This commit is contained in:
Karim shoair
2026-05-11 04:59:43 +03:00
committed by GitHub
46 changed files with 2271 additions and 488 deletions
+1 -1
View File
@@ -73,7 +73,7 @@ jobs:
- name: Install all browsers dependencies
run: |
python3 -m pip install --upgrade pip
python3 -m pip install playwright==1.58.0 patchright==1.58.2
python3 -m pip install playwright==1.59.0 patchright==1.59.1
- name: Get Playwright version
id: playwright-version
Binary file not shown.
+21 -2
View File
@@ -1,7 +1,7 @@
---
name: scrapling-official
description: Scrape web pages using Scrapling with anti-bot bypass (like Cloudflare Turnstile), stealth headless browsing, spiders framework, adaptive scraping, and JavaScript rendering. Use when asked to scrape, crawl, or extract data from websites; web_fetch fails; the site has anti-bot protections; write Python code to scrape/crawl; or write spiders.
version: "0.4.7"
version: "0.4.8"
license: Complete terms in LICENSE.txt
metadata:
homepage: "https://scrapling.readthedocs.io/en/latest/index.html"
@@ -40,7 +40,7 @@ Blazing fast crawls with real-time stats and streaming. Built by Web Scrapers fo
Create a virtual Python environment through any way available, like `venv`, then inside the environment do:
`pip install "scrapling[all]>=0.4.7"`
`pip install "scrapling[all]>=0.4.8"`
Then do this to download all the browsers' dependencies:
@@ -306,6 +306,25 @@ Press Ctrl+C to pause gracefully - progress is saved automatically. Later, when
While iterating on a spider's `parse()` logic, set `development_mode = True` on the spider class to cache responses to disk on the first run and replay them on subsequent runs - so you can re-run the spider as many times as you want without re-hitting the target servers. The cache lives in `.scrapling_cache/{spider.name}/` by default and can be overridden with `development_cache_dir`. Don't ship a spider with this enabled.
For rules-based crawls (follow links matching a regex), use `CrawlSpider` instead of writing the link-extraction loop yourself:
```python
from scrapling.spiders import CrawlSpider, CrawlRule, LinkExtractor
class BlogCrawler(CrawlSpider):
name = "blog"
start_urls = ["https://example.com"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post),
CrawlRule(LinkExtractor(allow=r"/page/\d+/")), # follow pagination, no callback
]
async def parse_post(self, response):
yield {"title": response.css("h1::text").get()}
```
For sitemap-driven crawls, use `SitemapSpider` with the same `rules()` API. It fetches `sitemap_urls`, descends into sitemap indexes, and dispatches each URL through your rules. Put a `robots.txt` URL directly in `sitemap_urls` and the spider extracts each `Sitemap:` directive from it automatically. See `references/spiders/generic-templates.md` for the full reference, including `LinkExtractor`'s allow/deny/restrict_css/canonicalize options.
### Advanced Parsing & Navigation
```python
from scrapling.fetchers import Fetcher
@@ -9,7 +9,7 @@ All examples collect **all 100 quotes across 10 pages**.
Make sure Scrapling is installed:
```bash
pip install "scrapling[all]>=0.4.7"
pip install "scrapling[all]>=0.4.8"
scrapling install --force
```
@@ -27,23 +27,23 @@ The following table compares them and can be quickly used for guidance.
## Parser configuration in all fetchers
All fetchers share the same import method, as you will see in the upcoming pages
```python
>>> from scrapling.fetchers import Fetcher, AsyncFetcher, StealthyFetcher, DynamicFetcher
from scrapling.fetchers import Fetcher, AsyncFetcher, StealthyFetcher, DynamicFetcher
```
Then you use it right away without initializing like this, and it will use the default parser settings:
```python
>>> page = StealthyFetcher.fetch('https://example.com')
page = StealthyFetcher.fetch('https://example.com')
```
If you want to configure the parser ([Selector class](parsing/main_classes.md#selector)) that will be used on the response before returning it for you, then do this first:
```python
>>> from scrapling.fetchers import Fetcher
>>> Fetcher.configure(adaptive=True, keep_comments=False, keep_cdata=False) # and the rest
from scrapling.fetchers import Fetcher
Fetcher.configure(adaptive=True, keep_comments=False, keep_cdata=False) # and the rest
```
or
```python
>>> from scrapling.fetchers import Fetcher
>>> Fetcher.adaptive=True
>>> Fetcher.keep_comments=False
>>> Fetcher.keep_cdata=False # and the rest
from scrapling.fetchers import Fetcher
Fetcher.adaptive=True
Fetcher.keep_comments=False
Fetcher.keep_cdata=False # and the rest
```
Then, continue your code as usual.
@@ -59,19 +59,19 @@ If your use case requires a different configuration for each request/fetch, you
## Response Object
The `Response` object is the same as the [Selector](parsing/main_classes.md#selector) class, but it has additional details about the response, like response headers, status, cookies, etc., as shown below:
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.get('https://example.com')
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://example.com')
>>> page.status # HTTP status code
>>> page.reason # Status message
>>> page.cookies # Response cookies as a dictionary
>>> page.headers # Response headers
>>> page.request_headers # Request headers
>>> page.history # Response history of redirections, if any
>>> page.body # Raw response body as bytes
>>> page.encoding # Response encoding
>>> page.meta # Response metadata dictionary (e.g., proxy used). Mainly helpful with the spiders system.
>>> page.captured_xhr # List of captured XHR/fetch responses (when capture_xhr is enabled on a browser session)
page.status # HTTP status code
page.reason # Status message
page.cookies # Response cookies as a dictionary
page.headers # Response headers
page.request_headers # Request headers
page.history # Response history of redirections, if any
page.body # Raw response body as bytes
page.encoding # Response encoding
page.meta # Response metadata dictionary (e.g., proxy used). Mainly helpful with the spiders system.
page.captured_xhr # List of captured XHR/fetch responses (when capture_xhr is enabled on a browser session)
```
All fetchers return the `Response` object.
@@ -8,7 +8,7 @@ As we will explain later, to automate the page, you need some knowledge of [Play
You have one primary way to import this Fetcher, which is the same for all fetchers.
```python
>>> from scrapling.fetchers import DynamicFetcher
from scrapling.fetchers import DynamicFetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -149,7 +149,7 @@ with DynamicSession(proxy_rotator=rotator, headless=True) as session:
### Downloading Files
```python
page = DynamicFetcher.fetch('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/images/main_cover.png')
page = DynamicFetcher.fetch('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')
with open(file='main_cover.png', mode='wb') as f:
f.write(page.body)
@@ -6,7 +6,7 @@ The `Fetcher` class provides rapid and lightweight HTTP requests using the high-
Import the Fetcher (same import pattern for all fetchers):
```python
>>> from scrapling.fetchers import Fetcher
from scrapling.fetchers import Fetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -47,41 +47,41 @@ Examples are the best way to explain this:
> Hence: `OPTIONS` and `HEAD` methods are not supported.
#### GET
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic GET
>>> page = Fetcher.get('https://example.com')
>>> page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = Fetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
>>> # With parameters
>>> page = Fetcher.get('https://example.com/search', params={'q': 'query'})
>>>
>>> # With headers
>>> page = Fetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
>>> # Basic HTTP authentication
>>> page = Fetcher.get("https://example.com", auth=("my_user", "password123"))
>>> # Browser impersonation
>>> page = Fetcher.get('https://example.com', impersonate='chrome')
>>> # HTTP/3 support
>>> page = Fetcher.get('https://example.com', http3=True)
from scrapling.fetchers import Fetcher
# Basic GET
page = Fetcher.get('https://example.com')
page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = Fetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = Fetcher.get('https://example.com/search', params={'q': 'query'})
# With headers
page = Fetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = Fetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = Fetcher.get('https://example.com', impersonate='chrome')
# HTTP/3 support
page = Fetcher.get('https://example.com', http3=True)
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic GET
>>> page = await AsyncFetcher.get('https://example.com')
>>> page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
>>> # With parameters
>>> page = await AsyncFetcher.get('https://example.com/search', params={'q': 'query'})
>>>
>>> # With headers
>>> page = await AsyncFetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
>>> # Basic HTTP authentication
>>> page = await AsyncFetcher.get("https://example.com", auth=("my_user", "password123"))
>>> # Browser impersonation
>>> page = await AsyncFetcher.get('https://example.com', impersonate='chrome110')
>>> # HTTP/3 support
>>> page = await AsyncFetcher.get('https://example.com', http3=True)
from scrapling.fetchers import AsyncFetcher
# Basic GET
page = await AsyncFetcher.get('https://example.com')
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = await AsyncFetcher.get('https://example.com/search', params={'q': 'query'})
# With headers
page = await AsyncFetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = await AsyncFetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = await AsyncFetcher.get('https://example.com', impersonate='chrome110')
# HTTP/3 support
page = await AsyncFetcher.get('https://example.com', http3=True)
```
The `page` object in all cases is a [Response](choosing.md#response-object) object, which is a [Selector](parsing/main_classes.md#selector), so you can use it directly
```python
@@ -102,62 +102,62 @@ The `page` object in all cases is a [Response](choosing.md#response-object) obje
```
#### POST
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic POST
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, params={'q': 'query'})
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
>>> # Another example of form-encoded data
>>> page = Fetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
>>> # JSON data
>>> page = Fetcher.post('https://example.com/api', json={'key': 'value'})
from scrapling.fetchers import Fetcher
# Basic POST
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, params={'q': 'query'})
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = Fetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = Fetcher.post('https://example.com/api', json={'key': 'value'})
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic POST
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
>>> # Another example of form-encoded data
>>> page = await AsyncFetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
>>> # JSON data
>>> page = await AsyncFetcher.post('https://example.com/api', json={'key': 'value'})
from scrapling.fetchers import AsyncFetcher
# Basic POST
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = await AsyncFetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = await AsyncFetcher.post('https://example.com/api', json={'key': 'value'})
```
#### PUT
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic PUT
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'})
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
>>> # Another example of form-encoded data
>>> page = Fetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
from scrapling.fetchers import Fetcher
# Basic PUT
page = Fetcher.put('https://example.com/update', data={'status': 'updated'})
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = Fetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic PUT
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'})
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
>>> # Another example of form-encoded data
>>> page = await AsyncFetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
from scrapling.fetchers import AsyncFetcher
# Basic PUT
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'})
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = await AsyncFetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
```
#### DELETE
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.delete('https://example.com/resource/123')
>>> page = Fetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
>>> page = Fetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
from scrapling.fetchers import Fetcher
page = Fetcher.delete('https://example.com/resource/123')
page = Fetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = Fetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> page = await AsyncFetcher.delete('https://example.com/resource/123')
>>> page = await AsyncFetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
>>> page = await AsyncFetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
from scrapling.fetchers import AsyncFetcher
page = await AsyncFetcher.delete('https://example.com/resource/123')
page = await AsyncFetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
```
## Session Management
@@ -301,7 +301,7 @@ def scrape_products():
```python
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/images/main_cover.png')
page = Fetcher.get('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')
with open(file='main_cover.png', mode='wb') as f:
f.write(page.body)
```
@@ -6,7 +6,7 @@
You have one primary way to import this Fetcher, which is the same for all fetchers.
```python
>>> from scrapling.fetchers import StealthyFetcher
from scrapling.fetchers import StealthyFetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -68,22 +68,21 @@ To extract the Questions button from the old design, a selector like `#hmenus >
Testing the same selector in both versions:
```python
>> from scrapling import Fetcher
>> selector = '#hmenus > div:nth-child(1) > ul > li:nth-child(1) > a'
>> old_url = "https://web.archive.org/web/20100102003420/http://stackoverflow.com/"
>> new_url = "https://stackoverflow.com/"
>> Fetcher.configure(adaptive = True, adaptive_domain='stackoverflow.com')
>>
>> page = Fetcher.get(old_url, timeout=30)
>> element1 = page.css(selector, auto_save=True)[0]
>>
>> # Same selector but used in the updated website
>> page = Fetcher.get(new_url)
>> element2 = page.css(selector, adaptive=True)[0]
>>
>> if element1.text == element2.text:
from scrapling import Fetcher
selector = '#hmenus > div:nth-child(1) > ul > li:nth-child(1) > a'
old_url = "https://web.archive.org/web/20100102003420/http://stackoverflow.com/"
new_url = "https://stackoverflow.com/"
Fetcher.configure(adaptive = True, adaptive_domain='stackoverflow.com')
page = Fetcher.get(old_url, timeout=30)
element1 = page.css(selector, auto_save=True)[0]
# Same selector but used in the updated website
page = Fetcher.get(new_url)
element2 = page.css(selector, adaptive=True)[0]
if element1.text == element2.text:
... print('Scrapling found the same element in the old and new designs!')
'Scrapling found the same element in the old and new designs!'
```
The `adaptive_domain` argument is used here because Scrapling sees `archive.org` and `stackoverflow.com` as two different domains and would isolate their `adaptive` data. Passing `adaptive_domain` tells Scrapling to treat them as the same website for adaptive data storage.
@@ -127,11 +126,11 @@ First, enable the `adaptive` feature by passing `adaptive=True` to the [Selector
Examples:
```python
>>> from scrapling import Selector, Fetcher
>>> page = Selector(html_doc, adaptive=True)
from scrapling import Selector, Fetcher
page = Selector(html_doc, adaptive=True)
# OR
>>> Fetcher.adaptive = True
>>> page = Fetcher.get('https://example.com')
Fetcher.adaptive = True
page = Fetcher.get('https://example.com')
```
When using the [Selector](main_classes.md#selector) class, pass the URL of the website with the `url` argument so Scrapling can separate the properties saved for each element by domain.
@@ -159,11 +158,11 @@ Elements can be manually saved, retrieved, and relocated within the `adaptive` f
Example of getting an element by text:
```python
>>> element = page.find_by_text('Tipping the Velvet', first_match=True)
element = page.find_by_text('Tipping the Velvet', first_match=True)
```
Save its unique properties using the `save` method. The identifier must be set manually (use a meaningful identifier):
```python
>>> page.save(element, 'my_special_element')
page.save(element, 'my_special_element')
```
Later, retrieve and relocate the element inside the page with `adaptive`:
```python
@@ -131,14 +131,14 @@ Getting the attributes of the element
```
Access a specific attribute with any of the following
```python
>>> article.attrib['class']
>>> article.attrib.get('class')
>>> article['class'] # new in v0.3
article.attrib['class']
article.attrib.get('class')
article['class'] # new in v0.3
```
Check if the attributes contain a specific attribute with any of the methods below
```python
>>> 'class' in article.attrib
>>> 'class' in article # new in v0.3
'class' in article.attrib
'class' in article # new in v0.3
```
Get the HTML content of the element
```python
@@ -279,13 +279,13 @@ In the [Selector](#selector) class, all methods/properties that should return a
Starting with v0.4, all selection methods consistently return [Selector](#selector)/[Selectors](#selectors) objects, even for text nodes and attribute values. Text nodes (selected via `::text`, `/text()`, `::attr()`, `/@attr`) are wrapped in [Selector](#selector) objects. These text node selectors have `tag` set to `"#text"`, and their `text` property returns the text value. You can still access the text value directly, and all other properties return empty/default values gracefully.
```python
>>> page.css('a::text') # -> Selectors (of text node Selectors)
>>> page.xpath('//a/text()') # -> Selectors
>>> page.css('a::text').get() # -> TextHandler (the first text value)
>>> page.css('a::text').getall() # -> TextHandlers (all text values)
>>> page.css('a::attr(href)') # -> Selectors
>>> page.xpath('//a/@href') # -> Selectors
>>> page.css('.price_color') # -> Selectors
page.css('a::text') # -> Selectors (of text node Selectors)
page.xpath('//a/text()') # -> Selectors
page.css('a::text').get() # -> TextHandler (the first text value)
page.css('a::text').getall() # -> TextHandlers (all text values)
page.css('a::attr(href)') # -> Selectors
page.xpath('//a/@href') # -> Selectors
page.css('.price_color') # -> Selectors
```
### Data extraction methods
@@ -346,8 +346,8 @@ It filters all elements in the current page/element in the following order:
### Examples
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.get('https://quotes.toscrape.com/')
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://quotes.toscrape.com/')
```
Find all elements with the tag name `div`.
```python
@@ -0,0 +1,167 @@
# Generic Spider Templates
Most crawls fall into one of two patterns: "follow links matching this regex" or "crawl every URL listed in the site's sitemap". Scrapling ships templates for both so you don't have to hand-write the same `parse()` boilerplate every time.
Both templates build on `LinkExtractor`, which pulls URLs out of a `Response` (or filters a single URL via `matches()`). `SitemapSpider` additionally parses sitemap.xml / sitemap_index.xml bodies internally (gzip-compressed or not).
You can use `LinkExtractor` directly inside any plain `Spider.parse()`. The templates just save you the wiring.
## CrawlSpider
`CrawlSpider` follows links automatically based on declarative rules.
```python
from scrapling.spiders import CrawlSpider, CrawlRule, LinkExtractor
class BlogCrawler(CrawlSpider):
name = "blog"
start_urls = ["https://example.com"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post),
CrawlRule(LinkExtractor(allow=r"/page/\d+/")), # follow pagination, no callback
]
async def parse_post(self, response):
yield {
"title": response.css("h1::text").get(),
"url": response.url,
}
result = BlogCrawler().start()
```
A `CrawlRule` pairs a `LinkExtractor` with an optional `callback` (a bound method on the spider), an optional `priority` override for the dispatched `Request`, and an optional `process_request` (a bound method that mutates each `Request` before it's yielded). The default `parse()` runs every rule against every response and yields a `Request` per matched URL.
If a rule has no callback, the matched URLs fall through to the spider's default `parse()` (or stay uncallback'd if you didn't override it). This is convenient for pagination: extract the next-page links to keep the crawl going, but don't need a separate handler.
### Combining rules with custom logic
Override `parse()` and call `super().parse(response)` to get the rule behavior plus your own yields:
```python
class MySpider(CrawlSpider):
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
async def parse(self, response):
yield {"page_url": response.url}
async for req in super().parse(response):
yield req
```
### Mutating Requests with `process_request`
```python
def add_priority(self, request, response):
request.priority = 10
return request
def rules(self):
return [CrawlRule(
LinkExtractor(allow=r"/posts/"),
callback=self.parse_post,
process_request=self.add_priority,
)]
```
## SitemapSpider
`SitemapSpider` seeds a crawl from sitemap.xml URLs. It uses the same `rules()` API as `CrawlSpider`, so the mental model is shared.
```python
from scrapling.spiders import SitemapSpider, CrawlRule, LinkExtractor
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/sitemap.xml"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post),
CrawlRule(LinkExtractor(allow=r"/products/"), callback=self.parse_product),
]
async def parse_post(self, response):
yield {"title": response.css("h1::text").get()}
async def parse_product(self, response):
yield {"sku": response.css(".sku::text").get()}
result = MySitemap().start()
```
### How URLs are dispatched
For each URL in the sitemap, `SitemapSpider` checks every rule's `LinkExtractor.matches(url)` in order. The first matching rule wins, and a `Request` is yielded with that rule's callback. If no rule matches and `rules()` is non-empty, the URL is dropped (matches Scrapy's behavior). If `rules()` returns an empty list, every URL is routed to the spider's `parse()` method, which raises `NotImplementedError` by default - override it to handle them.
### Sitemap indexes
When `SitemapSpider` encounters a `<sitemapindex>` (a sitemap of sitemaps), it descends into each child sitemap automatically. To filter which child sitemaps to descend into, set `sitemap_follow` to a `LinkExtractor`:
```python
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/sitemap.xml"]
sitemap_follow = LinkExtractor(allow=r"/posts-sitemap-\d+\.xml") # only post sitemaps
```
### Robots.txt support
Put a `robots.txt` URL directly in `sitemap_urls` and `SitemapSpider` will detect it, extract every `Sitemap:` directive (via `protego`), and follow each one:
```python
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/robots.txt"] # Sitemap: directives discovered automatically
```
### Alternate-language URLs
Set `sitemap_alternate_links = True` to also dispatch `<xhtml:link rel="alternate" hreflang="...">` URLs through your rules.
## Using `LinkExtractor` directly
You don't have to use the templates. `LinkExtractor` works inside any plain `Spider`:
```python
from scrapling.spiders import Spider, LinkExtractor
class CustomSpider(Spider):
name = "custom"
start_urls = ["https://example.com"]
def __init__(self):
super().__init__()
self._links = LinkExtractor(allow=r"/posts/", deny_domains="ads.example.com")
async def parse(self, response):
for url in self._links.extract(response):
yield response.follow(url, callback=self.parse_post)
async def parse_post(self, response):
yield {"title": response.css("h1::text").get()}
```
## LinkExtractor reference
| Argument | Default | Description |
|---|---|---|
| `allow` | `()` | URL patterns to keep. Empty means "match all". String, compiled `Pattern`, or iterable of either. |
| `deny` | `()` | URL patterns to drop. Always overrides `allow`. |
| `allow_domains` | `()` | Hostnames to keep. Subdomains match automatically (`example.com` matches `api.example.com`). |
| `deny_domains` | `()` | Hostnames to drop. |
| `restrict_css` | `()` | CSS selectors that scope DOM extraction to a region. |
| `restrict_xpath` | `()` | XPath selectors that scope DOM extraction to a region. |
| `tags` | `("a", "area")` | Element tags to look for links in. |
| `attrs` | `("href",)` | Attributes on those tags to read URLs from. |
| `canonicalize` | `True` | Sort query params and normalize the path. |
| `strip` | `True` | Strip whitespace from extracted URLs. |
| `keep_fragment` | `False` | Preserve the `#fragment` when canonicalizing. |
| `deny_extensions` | `IGNORED_EXTENSIONS` | File extensions to drop (pdf, zip, images, video, etc.). |
| `process` | `None` | Optional callable applied to each extracted URL before filtering. Return a falsy value to drop. |
`LinkExtractor.extract(response)` returns a `list[str]` of absolute, filtered, deduped URLs.
`LinkExtractor.matches(url)` returns a `bool` - the URL-only filter (allow/deny/domain/extension), used by `SitemapSpider` to dispatch URLs without a `Response`.
+54 -65
View File
@@ -43,25 +43,24 @@ Once launched, you'll see the Scrapling banner and can immediately start scrapin
```python
# No imports needed - everything is ready!
>>> get('https://news.ycombinator.com')
get('https://news.ycombinator.com')
>>> # Explore the page structure
>>> page.css('a')[:5] # Look at first 5 links
# Explore the page structure
page.css('a')[:5] # Look at first 5 links
>>> # Refine your selectors
>>> stories = page.css('.titleline>a')
>>> len(stories)
30
# Refine your selectors
stories = page.css('.titleline>a')
len(stories) # 30
>>> # Extract specific data
>>> for story in stories[:3]:
# Extract specific data
for story in stories[:3]:
... title = story.text
... url = story['href']
... print(f"{title}: {url}")
>>> # Try different approaches
>>> titles = page.css('.titleline>a::text') # Direct text extraction
>>> urls = page.css('.titleline>a::attr(href)') # Direct attribute extraction
# Try different approaches
titles = page.css('.titleline>a::text') # Direct text extraction
urls = page.css('.titleline>a::attr(href)') # Direct attribute extraction
```
## Built-in Shortcuts
@@ -86,12 +85,10 @@ The shell automatically tracks your requests and pages:
The `page` and `response` commands are automatically updated with the last fetched page:
```python
>>> get('https://quotes.toscrape.com')
>>> # 'page' and 'response' both refer to the last fetched page
>>> page.url
'https://quotes.toscrape.com'
>>> response.status # Same as page.status
200
get('https://quotes.toscrape.com')
# 'page' and 'response' both refer to the last fetched page
page.url # 'https://quotes.toscrape.com'
response.status # Prints 200; Same as page.status
```
- **Page History**
@@ -99,20 +96,17 @@ The shell automatically tracks your requests and pages:
The `pages` command keeps track of the last five pages (it's a `Selectors` object):
```python
>>> get('https://site1.com')
>>> get('https://site2.com')
>>> get('https://site3.com')
get('https://site1.com')
get('https://site2.com')
get('https://site3.com')
>>> # Access last 5 pages
>>> len(pages) # `Selectors` object with `page` history
3
>>> pages[0].url # First page in history
'https://site1.com'
>>> pages[-1].url # Most recent page
'https://site3.com'
# Access last 5 pages
len(pages) # `Selectors` object with `page` history -> 3
pages[0].url # First page in history -> 'https://site1.com'
pages[-1].url # Most recent page -> 'https://site3.com'
>>> # Work with historical pages
>>> for i, old_page in enumerate(pages):
# Work with historical pages
for i, old_page in enumerate(pages):
... print(f"Page {i}: {old_page.url} - {old_page.status}")
```
@@ -123,8 +117,8 @@ The shell automatically tracks your requests and pages:
View scraped pages in your browser:
```python
>>> get('https://quotes.toscrape.com')
>>> view(page) # Opens the page HTML in your default browser
get('https://quotes.toscrape.com')
view(page) # Opens the page HTML in your default browser
```
### Curl Command Integration
@@ -138,29 +132,24 @@ First, you need to copy a request as a curl command like the following:
- **Convert Curl command to Request Object**
```python
>>> curl_cmd = '''curl 'https://scrapling.requestcatcher.com/post' \
curl_cmd = '''curl 'https://scrapling.requestcatcher.com/post' \
... -X POST \
... -H 'Content-Type: application/json' \
... -d '{"name": "test", "value": 123}' '''
>>> request = uncurl(curl_cmd)
>>> request.method
'post'
>>> request.url
'https://scrapling.requestcatcher.com/post'
>>> request.headers
{'Content-Type': 'application/json'}
request = uncurl(curl_cmd)
request.method # -> 'post'
request.url # -> 'https://scrapling.requestcatcher.com/post'
request.headers # -> {'Content-Type': 'application/json'}
```
- **Execute Curl Command Directly**
```python
>>> # Convert and execute in one step
>>> curl2fetcher(curl_cmd)
>>> page.status
200
>>> page.json()['json']
{'name': 'test', 'value': 123}
# Convert and execute in one step
curl2fetcher(curl_cmd)
page.status # -> 200
page.json()['json'] # -> {'name': 'test', 'value': 123}
```
### IPython Features
@@ -168,17 +157,17 @@ First, you need to copy a request as a curl command like the following:
The shell inherits all IPython capabilities:
```python
>>> # Magic commands
>>> %time page = get('https://example.com') # Time execution
>>> %history # Show command history
>>> %save filename.py 1-10 # Save commands 1-10 to file
# Magic commands
%time page = get('https://example.com') # Time execution
%history # Show command history
%save filename.py 1-10 # Save commands 1-10 to file
>>> # Tab completion works everywhere
>>> page.c<TAB> # Shows: css, cookies, headers, etc.
>>> Fetcher.<TAB> # Shows all Fetcher methods
# Tab completion works everywhere
page.c<TAB> # Shows: css, cookies, headers, etc.
Fetcher.<TAB> # Shows all Fetcher methods
>>> # Object inspection
>>> get? # Show get documentation
# Object inspection
get? # Show get documentation
```
## Examples
@@ -188,23 +177,23 @@ Here are a few examples generated via AI:
#### E-commerce Data Collection
```python
>>> # Start with product listing page
>>> catalog = get('https://shop.example.com/products')
# Start with product listing page
catalog = get('https://shop.example.com/products')
>>> # Find product links
>>> product_links = catalog.css('.product-link::attr(href)')
>>> print(f"Found {len(product_links)} products")
# Find product links
product_links = catalog.css('.product-link::attr(href)')
print(f"Found {len(product_links)} products")
>>> # Sample a few products first
>>> for link in product_links[:3]:
# Sample a few products first
for link in product_links[:3]:
... product = get(f"https://shop.example.com{link}")
... name = product.css('.product-name::text').get('')
... price = product.css('.price::text').get('')
... print(f"{name}: {price}")
>>> # Scale up with sessions for efficiency
>>> from scrapling.fetchers import FetcherSession
>>> with FetcherSession() as session:
# Scale up with sessions for efficiency
from scrapling.fetchers import FetcherSession
with FetcherSession() as session:
... products = []
... for link in product_links:
... product = session.get(f"https://shop.example.com{link}")
+5 -6
View File
@@ -4,13 +4,12 @@
### All current types can be imported alone, like below
```python
>>> from scrapling.core.custom_types import TextHandler, AttributesHandler
from scrapling.core.custom_types import TextHandler, AttributesHandler
>>> somestring = TextHandler('{}')
>>> somestring.json()
'{}'
>>> somedict_1 = AttributesHandler({'a': 1})
>>> somedict_2 = AttributesHandler(a=1)
somestring = TextHandler('{}')
somestring.json() # '{}'
somedict_1 = AttributesHandler({'a': 1})
somedict_2 = AttributesHandler(a=1)
```
Note that `TextHandler` is a subclass of Python's `str`, so all standard operations/methods that work with Python strings will work.
+20 -20
View File
@@ -31,23 +31,23 @@ In the following pages, we will talk about each one in detail.
## Parser configuration in all fetchers
All fetchers share the same import method, as you will see in the upcoming pages
```python
>>> from scrapling.fetchers import Fetcher, AsyncFetcher, StealthyFetcher, DynamicFetcher
from scrapling.fetchers import Fetcher, AsyncFetcher, StealthyFetcher, DynamicFetcher
```
Then you use it right away without initializing like this, and it will use the default parser settings:
```python
>>> page = StealthyFetcher.fetch('https://example.com')
page = StealthyFetcher.fetch('https://example.com')
```
If you want to configure the parser ([Selector class](../parsing/main_classes.md#selector)) that will be used on the response before returning it for you, then do this first:
```python
>>> from scrapling.fetchers import Fetcher
>>> Fetcher.configure(adaptive=True, keep_comments=False, keep_cdata=False) # and the rest
from scrapling.fetchers import Fetcher
Fetcher.configure(adaptive=True, keep_comments=False, keep_cdata=False) # and the rest
```
or
```python
>>> from scrapling.fetchers import Fetcher
>>> Fetcher.adaptive=True
>>> Fetcher.keep_comments=False
>>> Fetcher.keep_cdata=False # and the rest
from scrapling.fetchers import Fetcher
Fetcher.adaptive=True
Fetcher.keep_comments=False
Fetcher.keep_cdata=False # and the rest
```
Then, continue your code as usual.
@@ -65,19 +65,19 @@ If your use case requires a different configuration for each request/fetch, you
## Response Object
The `Response` object is the same as the [Selector](../parsing/main_classes.md#selector) class, but it has additional details about the response, like response headers, status, cookies, etc., as shown below:
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.get('https://example.com')
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://example.com')
>>> page.status # HTTP status code
>>> page.reason # Status message
>>> page.cookies # Response cookies as a dictionary
>>> page.headers # Response headers
>>> page.request_headers # Request headers
>>> page.history # Response history of redirections, if any
>>> page.body # Raw response body as bytes
>>> page.encoding # Response encoding
>>> page.meta # Response metadata dictionary (e.g., proxy used). Mainly helpful with the spiders system.
>>> page.captured_xhr # List of captured XHR/fetch responses (when capture_xhr is enabled on a browser session)
page.status # HTTP status code
page.reason # Status message
page.cookies # Response cookies as a dictionary
page.headers # Response headers
page.request_headers # Request headers
page.history # Response history of redirections, if any
page.body # Raw response body as bytes
page.encoding # Response encoding
page.meta # Response metadata dictionary (e.g., proxy used). Mainly helpful with the spiders system.
page.captured_xhr # List of captured XHR/fetch responses (when capture_xhr is enabled on a browser session)
```
All fetchers return the `Response` object.
+82 -81
View File
@@ -14,7 +14,7 @@ As we will explain later, to automate the page, you need some knowledge of [Play
You have one primary way to import this Fetcher, which is the same for all fetchers.
```python
>>> from scrapling.fetchers import DynamicFetcher
from scrapling.fetchers import DynamicFetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -77,7 +77,7 @@ Scrapling provides many options with this fetcher and its session classes. To ma
| wait_selector | Wait for a specific css selector to be in a specific state. | ✔️ |
| init_script | An absolute path to a JavaScript file to be executed on page creation for all pages in this session. | ✔️ |
| wait_selector_state | Scrapling will wait for the given state to be fulfilled for the selector given with `wait_selector`. _Default state is `attached`._ | ✔️ |
| google_search | Enabled by default, Scrapling will set a Google referer header. | ✔️ |
| google_search | Enabled by default, Scrapling will set a Google referer header. | ✔️ |
| extra_headers | A dictionary of extra headers to add to the request. _The referer set by `google_search` takes priority over the referer set here if used together._ | ✔️ |
| proxy | The proxy to be used with requests. It can be a string or a dictionary with only the keys 'server', 'username', and 'password'. | ✔️ |
| real_chrome | If you have a Chrome browser installed on your device, enable this, and the Fetcher will launch and use an instance of your browser. | ✔️ |
@@ -89,12 +89,12 @@ Scrapling provides many options with this fetcher and its session classes. To ma
| additional_args | Additional arguments to be passed to Playwright's context as additional settings, and they take higher priority than Scrapling's settings. | ✔️ |
| selector_config | A dictionary of custom parsing arguments to be used when creating the final `Selector`/`Response` class. | ✔️ |
| blocked_domains | A set of domain names to block requests to. Subdomains are also matched (e.g., `"example.com"` blocks `"sub.example.com"` too). | ✔️ |
| block_ads | Block requests to ~3,500 known ad/tracking domains. Can be combined with `blocked_domains`. | ✔️ |
| block_ads | Block requests to ~3,500 known ad/tracking domains. Can be combined with `blocked_domains`. | ✔️ |
| dns_over_https | Route DNS queries through Cloudflare's DNS-over-HTTPS to prevent DNS leaks when using proxies. | ✔️ |
| proxy_rotator | A `ProxyRotator` instance for automatic proxy rotation. Cannot be combined with `proxy`. | ✔️ |
| retries | Number of retry attempts for failed requests. Defaults to 3. | ✔️ |
| retry_delay | Seconds to wait between retry attempts. Defaults to 1. | ✔️ |
| capture_xhr | Pass a regex URL pattern string to capture XHR/fetch requests matching it during page load. Captured responses are available via `response.captured_xhr`. Defaults to `None` (disabled). | ✔️ |
| capture_xhr | Pass a regex URL pattern string to capture XHR/fetch requests matching it during page load. Captured responses are available via `response.captured_xhr`. Defaults to `None` (disabled). | ✔️ |
| executable_path | Absolute path to a custom browser executable to use instead of the bundled Chromium. Useful for non-standard installations or custom browser builds. | ✔️ |
In session classes, all these arguments can be set globally for the session. Still, you can configure each request individually by passing some of the arguments here that can be configured on the browser tab level like: `google_search`, `timeout`, `wait`, `page_action`, `page_setup`, `extra_headers`, `disable_resources`, `wait_selector`, `wait_selector_state`, `network_idle`, `load_dom`, `blocked_domains`, `proxy`, and `selector_config`.
@@ -107,6 +107,65 @@ In session classes, all these arguments can be set globally for the session. Sti
4. If you didn't set a user agent and enabled headless mode, the fetcher will generate a real user agent for the same browser version and use it. If you didn't set a user agent and didn't enable headless mode, the fetcher will use the browser's default user agent, which is the same as in standard browsers in the latest versions.
## Session Management
To keep the browser open until you make multiple requests with the same configuration, use `DynamicSession`/`AsyncDynamicSession` classes. Those classes can accept all the arguments that the `fetch` function can take, which enables you to specify a config for the entire session.
```python
from scrapling.fetchers import DynamicSession
# Create a session with default configuration
with DynamicSession(
headless=True,
disable_resources=True,
real_chrome=True
) as session:
# Make multiple requests with the same browser instance
page1 = session.fetch('https://example1.com')
page2 = session.fetch('https://example2.com')
page3 = session.fetch('https://dynamic-site.com')
# All requests reuse the same tab on the same browser instance
```
### Async Session Usage
```python
import asyncio
from scrapling.fetchers import AsyncDynamicSession
async def scrape_multiple_sites():
async with AsyncDynamicSession(
network_idle=True,
timeout=30000,
max_pages=3
) as session:
# Make async requests with shared browser configuration
pages = await asyncio.gather(
session.fetch('https://spa-app1.com'),
session.fetch('https://spa-app2.com'),
session.fetch('https://dynamic-content.com')
)
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
### Session Benefits
- **Browser reuse**: Much faster subsequent requests by reusing the same browser instance.
- **Cookie persistence**: Automatic cookie and session state handling as any browser does automatically.
- **Consistent fingerprint**: Same browser fingerprint across all requests.
- **Memory efficiency**: Better resource usage compared to launching new browsers with each fetch.
## Examples
It's easier to understand with examples, so let's take a look.
@@ -137,35 +196,10 @@ page = DynamicFetcher.fetch('https://example.com', timeout=30000) # 30 seconds
page = DynamicFetcher.fetch('https://example.com', proxy='http://username:password@host:port')
```
### Proxy Rotation
```python
from scrapling.fetchers import DynamicSession, ProxyRotator
# Set up proxy rotation
rotator = ProxyRotator([
"http://proxy1:8080",
"http://proxy2:8080",
"http://proxy3:8080",
])
# Use with session - rotates proxy automatically with each request
with DynamicSession(proxy_rotator=rotator, headless=True) as session:
page1 = session.fetch('https://example1.com')
page2 = session.fetch('https://example2.com')
# Override rotator for a specific request
page3 = session.fetch('https://example3.com', proxy='http://specific-proxy:8080')
```
!!! warning
Remember that by default, all browser-based fetchers and sessions use a persistent browser context with a pool of tabs. However, since browsers can't set a proxy per tab, when you use a `ProxyRotator`, the fetcher will automatically open a separate context for each proxy, with one tab per context. Once the tab's job is done, both the tab and its context are closed.
### Downloading Files
```python
page = DynamicFetcher.fetch('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/images/main_cover.png')
page = DynamicFetcher.fetch('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')
with open(file='main_cover.png', mode='wb') as f:
f.write(page.body)
@@ -229,8 +263,8 @@ page = await DynamicFetcher.async_fetch('https://example.com', page_action=scrol
```python
# Wait for the selector
page = DynamicFetcher.fetch(
'https://example.com',
wait_selector='h1',
'https://quotes.toscrape.com/js-delayed/',
wait_selector='.quote',
wait_selector_state='visible'
)
```
@@ -297,63 +331,30 @@ def scrape_dynamic_content():
}
```
## Session Management
To keep the browser open until you make multiple requests with the same configuration, use `DynamicSession`/`AsyncDynamicSession` classes. Those classes can accept all the arguments that the `fetch` function can take, which enables you to specify a config for the entire session.
### Proxy Rotation
```python
from scrapling.fetchers import DynamicSession
from scrapling.fetchers import DynamicSession, ProxyRotator
# Create a session with default configuration
with DynamicSession(
headless=True,
disable_resources=True,
real_chrome=True
) as session:
# Make multiple requests with the same browser instance
# Set up proxy rotation
rotator = ProxyRotator([
"http://proxy1:8080",
"http://proxy2:8080",
"http://proxy3:8080",
])
# Use with session - rotates proxy automatically with each request
with DynamicSession(proxy_rotator=rotator, headless=True) as session:
page1 = session.fetch('https://example1.com')
page2 = session.fetch('https://example2.com')
page3 = session.fetch('https://dynamic-site.com')
# All requests reuse the same tab on the same browser instance
# Override rotator for a specific request
page3 = session.fetch('https://example3.com', proxy='http://specific-proxy:8080')
```
### Async Session Usage
!!! warning
```python
import asyncio
from scrapling.fetchers import AsyncDynamicSession
async def scrape_multiple_sites():
async with AsyncDynamicSession(
network_idle=True,
timeout=30000,
max_pages=3
) as session:
# Make async requests with shared browser configuration
pages = await asyncio.gather(
session.fetch('https://spa-app1.com'),
session.fetch('https://spa-app2.com'),
session.fetch('https://dynamic-content.com')
)
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
### Session Benefits
- **Browser reuse**: Much faster subsequent requests by reusing the same browser instance.
- **Cookie persistence**: Automatic cookie and session state handling as any browser does automatically.
- **Consistent fingerprint**: Same browser fingerprint across all requests.
- **Memory efficiency**: Better resource usage compared to launching new browsers with each fetch.
Remember that by default, all browser-based fetchers and sessions use a persistent browser context with a pool of tabs. However, since browsers can't set a proxy per tab, when you use a `ProxyRotator`, the fetcher will automatically open a separate context for each proxy, with one tab per context. Once the tab's job is done, both the tab and its context are closed.
## When to Use
+75 -75
View File
@@ -12,7 +12,7 @@ The `Fetcher` class provides rapid and lightweight HTTP requests using the high-
You have one primary way to import this Fetcher, which is the same for all fetchers.
```python
>>> from scrapling.fetchers import Fetcher
from scrapling.fetchers import Fetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -54,47 +54,47 @@ Examples are the best way to explain this:
> Hence: `OPTIONS` and `HEAD` methods are not supported.
#### GET
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic GET
>>> page = Fetcher.get('https://example.com')
>>> page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = Fetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
>>> # With parameters
>>> page = Fetcher.get('https://example.com/search', params={'q': 'query'})
>>>
>>> # With headers
>>> page = Fetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
>>> # Basic HTTP authentication
>>> page = Fetcher.get("https://example.com", auth=("my_user", "password123"))
>>> # Browser impersonation
>>> page = Fetcher.get('https://example.com', impersonate='chrome')
>>> # HTTP/3 support
>>> page = Fetcher.get('https://example.com', http3=True)
from scrapling.fetchers import Fetcher
# Basic GET
page = Fetcher.get('https://example.com')
page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = Fetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = Fetcher.get('https://example.com/search', params={'q': 'query'})
# With headers
page = Fetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = Fetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = Fetcher.get('https://example.com', impersonate='chrome')
# HTTP/3 support
page = Fetcher.get('https://example.com', http3=True)
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic GET
>>> page = await AsyncFetcher.get('https://example.com')
>>> page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
>>> # With parameters
>>> page = await AsyncFetcher.get('https://example.com/search', params={'q': 'query'})
from scrapling.fetchers import AsyncFetcher
# Basic GET
page = await AsyncFetcher.get('https://example.com')
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = await AsyncFetcher.get('https://example.com/search', params={'q': 'query'})
>>>
>>> # With headers
>>> page = await AsyncFetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
>>> # Basic HTTP authentication
>>> page = await AsyncFetcher.get("https://example.com", auth=("my_user", "password123"))
>>> # Browser impersonation
>>> page = await AsyncFetcher.get('https://example.com', impersonate='chrome110')
>>> # HTTP/3 support
>>> page = await AsyncFetcher.get('https://example.com', http3=True)
# With headers
page = await AsyncFetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = await AsyncFetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = await AsyncFetcher.get('https://example.com', impersonate='chrome110')
# HTTP/3 support
page = await AsyncFetcher.get('https://example.com', http3=True)
```
Needless to say, the `page` object in all cases is [Response](choosing.md#response-object) object, which is a [Selector](../parsing/main_classes.md#selector) as we said, so you can use it directly
```python
>>> page.css('.something.something')
page.css('.something.something')
>>> page = Fetcher.get('https://api.github.com/events')
page = Fetcher.get('https://api.github.com/events')
>>> page.json()
[{'id': '<redacted>',
'type': 'PushEvent',
@@ -109,62 +109,62 @@ Needless to say, the `page` object in all cases is [Response](choosing.md#respon
```
#### POST
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic POST
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, params={'q': 'query'})
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
>>> # Another example of form-encoded data
>>> page = Fetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
>>> # JSON data
>>> page = Fetcher.post('https://example.com/api', json={'key': 'value'})
from scrapling.fetchers import Fetcher
# Basic POST
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, params={'q': 'query'})
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = Fetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = Fetcher.post('https://example.com/api', json={'key': 'value'})
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic POST
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
>>> # Another example of form-encoded data
>>> page = await AsyncFetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
>>> # JSON data
>>> page = await AsyncFetcher.post('https://example.com/api', json={'key': 'value'})
from scrapling.fetchers import AsyncFetcher
# Basic POST
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = await AsyncFetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = await AsyncFetcher.post('https://example.com/api', json={'key': 'value'})
```
#### PUT
```python
>>> from scrapling.fetchers import Fetcher
>>> # Basic PUT
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'})
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
>>> page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
>>> # Another example of form-encoded data
>>> page = Fetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
from scrapling.fetchers import Fetcher
# Basic PUT
page = Fetcher.put('https://example.com/update', data={'status': 'updated'})
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = Fetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> # Basic PUT
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'})
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
>>> page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
>>> # Another example of form-encoded data
>>> page = await AsyncFetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
from scrapling.fetchers import AsyncFetcher
# Basic PUT
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'})
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = await AsyncFetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
```
#### DELETE
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.delete('https://example.com/resource/123')
>>> page = Fetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
>>> page = Fetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
from scrapling.fetchers import Fetcher
page = Fetcher.delete('https://example.com/resource/123')
page = Fetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = Fetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
```
And for asynchronous requests, it's a small adjustment
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> page = await AsyncFetcher.delete('https://example.com/resource/123')
>>> page = await AsyncFetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
>>> page = await AsyncFetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
from scrapling.fetchers import AsyncFetcher
page = await AsyncFetcher.delete('https://example.com/resource/123')
page = await AsyncFetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
```
## Session Management
@@ -308,7 +308,7 @@ def scrape_products():
```python
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/images/main_cover.png')
page = Fetcher.get('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')
with open(file='main_cover.png', mode='wb') as f:
f.write(page.body)
```
+6 -6
View File
@@ -15,7 +15,7 @@ As with [DynamicFetcher](dynamic.md#introduction), you will need some knowledge
You have one primary way to import this Fetcher, which is the same for all fetchers.
```python
>>> from scrapling.fetchers import StealthyFetcher
from scrapling.fetchers import StealthyFetcher
```
Check out how to configure the parsing options [here](choosing.md#parser-configuration-in-all-fetchers)
@@ -54,7 +54,7 @@ Scrapling provides many options with this fetcher and its session classes. Befor
| wait_selector | Wait for a specific css selector to be in a specific state. | ✔️ |
| init_script | An absolute path to a JavaScript file to be executed on page creation for all pages in this session. | ✔️ |
| wait_selector_state | Scrapling will wait for the given state to be fulfilled for the selector given with `wait_selector`. _Default state is `attached`._ | ✔️ |
| google_search | Enabled by default, Scrapling will set a Google referer header. | ✔️ |
| google_search | Enabled by default, Scrapling will set a Google referer header. | ✔️ |
| extra_headers | A dictionary of extra headers to add to the request. _The referer set by `google_search` takes priority over the referer set here if used together._ | ✔️ |
| proxy | The proxy to be used with requests. It can be a string or a dictionary with only the keys 'server', 'username', and 'password'. | ✔️ |
| real_chrome | If you have a Chrome browser installed on your device, enable this, and the Fetcher will launch and use an instance of your browser. | ✔️ |
@@ -70,12 +70,12 @@ Scrapling provides many options with this fetcher and its session classes. Befor
| additional_args | Additional arguments to be passed to Playwright's context as additional settings, and they take higher priority than Scrapling's settings. | ✔️ |
| selector_config | A dictionary of custom parsing arguments to be used when creating the final `Selector`/`Response` class. | ✔️ |
| blocked_domains | A set of domain names to block requests to. Subdomains are also matched (e.g., `"example.com"` blocks `"sub.example.com"` too). | ✔️ |
| block_ads | Block requests to ~3,500 known ad/tracking domains. Can be combined with `blocked_domains`. | ✔️ |
| block_ads | Block requests to ~3,500 known ad/tracking domains. Can be combined with `blocked_domains`. | ✔️ |
| dns_over_https | Route DNS queries through Cloudflare's DNS-over-HTTPS to prevent DNS leaks when using proxies. | ✔️ |
| proxy_rotator | A `ProxyRotator` instance for automatic proxy rotation. Cannot be combined with `proxy`. | ✔️ |
| retries | Number of retry attempts for failed requests. Defaults to 3. | ✔️ |
| retry_delay | Seconds to wait between retry attempts. Defaults to 1. | ✔️ |
| capture_xhr | Pass a regex URL pattern string to capture XHR/fetch requests matching it during page load. Captured responses are available via `response.captured_xhr`. Defaults to `None` (disabled). | ✔️ |
| capture_xhr | Pass a regex URL pattern string to capture XHR/fetch requests matching it during page load. Captured responses are available via `response.captured_xhr`. Defaults to `None` (disabled). | ✔️ |
| executable_path | Absolute path to a custom browser executable to use instead of the bundled Chromium. Useful for non-standard installations or custom browser builds. | ✔️ |
In session classes, all these arguments can be set globally for the session. Still, you can configure each request individually by passing some of the arguments here that can be configured on the browser tab level like: `google_search`, `timeout`, `wait`, `page_action`, `page_setup`, `extra_headers`, `disable_resources`, `wait_selector`, `wait_selector_state`, `network_idle`, `load_dom`, `solve_cloudflare`, `blocked_domains`, `proxy`, and `selector_config`.
@@ -154,8 +154,8 @@ page = await StealthyFetcher.async_fetch('https://example.com', page_action=scro
```python
# Wait for the selector
page = StealthyFetcher.fetch(
'https://example.com',
wait_selector='h1',
'https://quotes.toscrape.com/js-delayed/',
wait_selector='.quote',
wait_selector_state='visible'
)
```
+28 -30
View File
@@ -263,19 +263,19 @@ page = Fetcher.get('https://scrapling.requestcatcher.com/get', impersonate="chro
```
With that out of the way, here's how to do all HTTP methods:
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030')
>>> page = Fetcher.put('https://scrapling.requestcatcher.com/put', data={'key': 'value'})
>>> page = Fetcher.delete('https://scrapling.requestcatcher.com/delete')
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030')
page = Fetcher.put('https://scrapling.requestcatcher.com/put', data={'key': 'value'})
page = Fetcher.delete('https://scrapling.requestcatcher.com/delete')
```
For Async requests, you will replace the import like below:
```python
>>> from scrapling.fetchers import AsyncFetcher
>>> page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
>>> page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030')
>>> page = await AsyncFetcher.put('https://scrapling.requestcatcher.com/put', data={'key': 'value'})
>>> page = await AsyncFetcher.delete('https://scrapling.requestcatcher.com/delete')
from scrapling.fetchers import AsyncFetcher
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030')
page = await AsyncFetcher.put('https://scrapling.requestcatcher.com/put', data={'key': 'value'})
page = await AsyncFetcher.delete('https://scrapling.requestcatcher.com/delete')
```
!!! note "Notes:"
@@ -291,14 +291,13 @@ We have you covered if you deal with dynamic websites like most today!
The `DynamicFetcher` class (formerly `PlayWrightFetcher`) offers many options for fetching and loading web pages using Chromium-based browsers.
```python
>>> from scrapling.fetchers import DynamicFetcher
>>> page = DynamicFetcher.fetch('https://www.google.com/search?q=%22Scrapling%22', disable_resources=True) # Vanilla Playwright option
>>> page.css("#search a::attr(href)").get()
'https://github.com/D4Vinci/Scrapling'
>>> # The async version of fetch
>>> page = await DynamicFetcher.async_fetch('https://www.google.com/search?q=%22Scrapling%22', disable_resources=True)
>>> page.css("#search a::attr(href)").get()
'https://github.com/D4Vinci/Scrapling'
from scrapling.fetchers import DynamicFetcher
page = DynamicFetcher.fetch('https://quotes.toscrape.com/js/', disable_resources=True, block_ads=True)
print(len(page.css(".quote"))) # -> 10
# The async version of fetch
page = await DynamicFetcher.async_fetch('https://quotes.toscrape.com/js/', disable_resources=True, block_ads=True)
print(len(page.css(".quote"))) # -> 10
```
It's built on top of [Playwright](https://playwright.dev/python/), and it's currently providing two main run options that can be mixed as you want:
@@ -323,18 +322,17 @@ Some of the things it does:
6. and other anti-protection options...
```python
>>> from scrapling.fetchers import StealthyFetcher
>>> page = StealthyFetcher.fetch('https://www.browserscan.net/bot-detection') # Running headless by default
>>> page.status == 200
True
>>> page = StealthyFetcher.fetch('https://nopecha.com/demo/cloudflare', solve_cloudflare=True) # Solve Cloudflare captcha automatically if presented
>>> page.status == 200
True
>>> page = StealthyFetcher.fetch('https://www.browserscan.net/bot-detection', humanize=True, os_randomize=True) # and the rest of arguments...
>>> # The async version of fetch
>>> page = await StealthyFetcher.async_fetch('https://www.browserscan.net/bot-detection')
>>> page.status == 200
True
from scrapling.fetchers import StealthyFetcher
page = StealthyFetcher.fetch('https://www.browserscan.net/bot-detection') # Running headless by default
page.status == 200 # -> True
page = StealthyFetcher.fetch('https://nopecha.com/demo/cloudflare', solve_cloudflare=True) # Solve Cloudflare captcha automatically if presented
page.status == 200 # -> True
page = StealthyFetcher.fetch('https://www.browserscan.net/bot-detection', block_webrtc=True, hide_canvas=True, dns_over_https=True) # and the rest of arguments...
# The async version of fetch
page = await StealthyFetcher.async_fetch('https://www.browserscan.net/bot-detection')
page.status == 200 # -> True
```
Again, this is just the tip of the iceberg with this fetcher. Check out the rest from [here](fetching/stealthy.md) for all details and the complete list of arguments.
+19 -22
View File
@@ -76,22 +76,19 @@ If I want to extract the Questions button from the old design, I can use a selec
Now, let's test the same selector in both versions
```python
>> from scrapling import Fetcher
>> selector = '#hmenus > div:nth-child(1) > ul > li:nth-child(1) > a'
>> old_url = "https://web.archive.org/web/20100102003420/http://stackoverflow.com/"
>> new_url = "https://stackoverflow.com/"
>> Fetcher.configure(adaptive = True, adaptive_domain='stackoverflow.com')
>>
>> page = Fetcher.get(old_url, timeout=30)
>> element1 = page.css(selector, auto_save=True)[0]
>>
>> # Same selector but used in the updated website
>> page = Fetcher.get(new_url)
>> element2 = page.css(selector, adaptive=True)[0]
>>
>> if element1.text == element2.text:
... print('Scrapling found the same element in the old and new designs!')
'Scrapling found the same element in the old and new designs!'
from scrapling import Fetcher
selector = '#hmenus > div:nth-child(1) > ul > li:nth-child(1) > a'
old_url = "https://web.archive.org/web/20100102003420/http://stackoverflow.com/"
new_url = "https://stackoverflow.com/"
Fetcher.configure(adaptive = True, adaptive_domain='stackoverflow.com')
page = Fetcher.get(old_url, timeout=30)
element1 = page.css(selector, auto_save=True)[0]
# Same selector but used in the updated website
page = Fetcher.get(new_url)
element2 = page.css(selector, adaptive=True)[0]
if element1.text == element2.text:
print('Scrapling found the same element in the old and new designs!') # Spoiler alert: it does!
```
Note that I introduced a new argument called `adaptive_domain`. This is because, for Scrapling, these are two different domains (`archive.org` and `stackoverflow.com`), so Scrapling will isolate their `adaptive` data. To inform Scrapling that they are the same website, we must pass the custom domain we wish to use while saving `adaptive` data for both, ensuring Scrapling doesn't isolate them.
@@ -141,11 +138,11 @@ First, you must enable the `adaptive` feature by passing `adaptive=True` to the
Examples:
```python
>>> from scrapling import Selector, Fetcher
>>> page = Selector(html_doc, adaptive=True)
from scrapling import Selector, Fetcher
page = Selector(html_doc, adaptive=True)
# OR
>>> Fetcher.adaptive = True
>>> page = Fetcher.get('https://example.com')
Fetcher.adaptive = True
page = Fetcher.get('https://example.com')
```
If you are using the [Selector](main_classes.md#selector) class, you need to pass the url of the website you are using with the argument `url` so Scrapling can separate the properties saved for each element by domain.
@@ -175,11 +172,11 @@ You manually save and retrieve an element, then relocate it, which all happens w
First, let's say you got an element like this by text:
```python
>>> element = page.find_by_text('Tipping the Velvet', first_match=True)
element = page.find_by_text('Tipping the Velvet', first_match=True)
```
You can save its unique properties using the `save` method, as shown below, but you must set the identifier yourself. For this example, I chose `my_special_element` as an identifier, but it's best to use a meaningful identifier in your code for the same reason you use meaningful variable names :)
```python
>>> page.save(element, 'my_special_element')
page.save(element, 'my_special_element')
```
Now, later, when you want to retrieve it and relocate it inside the page with `adaptive`, it would be like this
```python
+12 -12
View File
@@ -140,14 +140,14 @@ Getting the attributes of the element
```
Access a specific attribute with any of the following
```python
>>> article.attrib['class']
>>> article.attrib.get('class')
>>> article['class'] # new in v0.3
article.attrib['class']
article.attrib.get('class')
article['class'] # new in v0.3
```
Check if the attributes contain a specific attribute with any of the methods below
```python
>>> 'class' in article.attrib
>>> 'class' in article # new in v0.3
'class' in article.attrib
'class' in article # new in v0.3
```
Get the HTML content of the element
```python
@@ -292,13 +292,13 @@ In the [Selector](#selector) class, all methods/properties that should return a
Starting with v0.4, all selection methods consistently return [Selector](#selector)/[Selectors](#selectors) objects, even for text nodes and attribute values. Text nodes (selected via `::text`, `/text()`, `::attr()`, `/@attr`) are wrapped in [Selector](#selector) objects. These text node selectors have `tag` set to `"#text"`, and their `text` property returns the text value. You can still access the text value directly, and all other properties return empty/default values gracefully.
```python
>>> page.css('a::text') # -> Selectors (of text node Selectors)
>>> page.xpath('//a/text()') # -> Selectors
>>> page.css('a::text').get() # -> TextHandler (the first text value)
>>> page.css('a::text').getall() # -> TextHandlers (all text values)
>>> page.css('a::attr(href)') # -> Selectors
>>> page.xpath('//a/@href') # -> Selectors
>>> page.css('.price_color') # -> Selectors
page.css('a::text') # -> Selectors (of text node Selectors)
page.xpath('//a/text()') # -> Selectors
page.css('a::text').get() # -> TextHandler (the first text value)
page.css('a::text').getall() # -> TextHandlers (all text values)
page.css('a::attr(href)') # -> Selectors
page.xpath('//a/@href') # -> Selectors
page.css('.price_color') # -> Selectors
```
### Data extraction methods
+2 -2
View File
@@ -362,8 +362,8 @@ Check examples to clear any confusion :)
### Examples
```python
>>> from scrapling.fetchers import Fetcher
>>> page = Fetcher.get('https://quotes.toscrape.com/')
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://quotes.toscrape.com/')
```
Find all elements with the tag name `div`.
```python
+2 -2
View File
@@ -1,5 +1,5 @@
zensical>=0.0.30
mkdocstrings>=1.0.3
zensical>=0.0.41
mkdocstrings>=1.0.4
mkdocstrings-python>=2.0.3
griffe-inherited-docstrings>=1.1.3
griffe-runtime-objects>=0.3.1
+168
View File
@@ -0,0 +1,168 @@
# Generic Spider Templates
Most crawls fall into one of two patterns: "follow links matching this pattern" or "crawl every URL listed in the site's sitemap". Scrapling ships templates for both so you don't have to hand-write the same `parse()` boilerplate every time.
All templates build on `LinkExtractor`, which pulls URLs out of a `Response` (or filters a single URL via `matches()`). `SitemapSpider` additionally parses sitemap.xml / sitemap_index.xml bodies internally (gzip-compressed or not).
You can use `LinkExtractor` directly inside any plain `Spider.parse()`. The templates just save you the wiring.
## CrawlSpider
`CrawlSpider` follows links automatically based on declarative rules.
```python
from scrapling.spiders import CrawlSpider, CrawlRule, LinkExtractor
class QuotesSpider(CrawlSpider):
name = "blog"
start_urls = ["https://quotes.toscrape.com/"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/author/"), callback=self.parse_author),
CrawlRule(LinkExtractor(allow=r"/page/\d+/")), # follow pagination, no callback
]
async def parse_author(self, response):
yield {
'.author-title': response.css('.author-title::text').get(),
"birthday": response.css('.author-born-date::text').get(),
"url": response.url,
}
result = QuotesSpider().start()
```
A `CrawlRule` pairs a `LinkExtractor` with an optional `callback` (a bound method on the spider), an optional `priority` override for the dispatched `Request`, and an optional `process_request` (a bound method that mutates each `Request` before it's yielded). The default `parse()` runs every rule against every response and yields a `Request` per matched URL.
If a rule has no callback, the matched URLs fall through to the spider's default `parse()`. This is convenient for pagination: extract the next-page links to keep the crawl going, without needing a separate handler.
### Combining rules with custom logic
Override `parse()` and call `super().parse(response)` to get the rule behavior plus your own yields:
```python
class MySpider(CrawlSpider):
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
async def parse(self, response):
yield {"page_url": response.url}
async for req in super().parse(response):
yield req
```
### Mutating Requests with `process_request`
```python
def add_priority(self, request, response):
request.priority = 10
return request
def rules(self):
return [CrawlRule(
LinkExtractor(allow=r"/posts/"),
callback=self.parse_post,
process_request=self.add_priority,
)]
```
## SitemapSpider
`SitemapSpider` seeds a crawl from sitemap.xml URLs. It uses the same `rules()` API as `CrawlSpider`, so the mental model is shared.
```python
from scrapling.spiders import SitemapSpider, CrawlRule, LinkExtractor
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/sitemap.xml"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post),
CrawlRule(LinkExtractor(allow=r"/products/"), callback=self.parse_product),
]
async def parse_post(self, response):
yield {"title": response.css("h1::text").get()}
async def parse_product(self, response):
yield {"sku": response.css(".sku::text").get()}
result = MySitemap().start()
```
### How URLs are dispatched
For each URL in the sitemap, `SitemapSpider` checks every rule's `LinkExtractor.matches(url)` in order. The first matching rule wins, and a `Request` is yielded with that rule's callback. If no rule matches and `rules()` is non-empty, the URL is dropped. If `rules()` returns an empty list, every URL is routed to the spider's `parse()` method, which raises `NotImplementedError` by default if not overridden.
### Sitemap indexes
When `SitemapSpider` encounters a `<sitemapindex>` (a sitemap of sitemaps), it descends into each child sitemap automatically. To filter which child sitemaps to descend into, set `sitemap_follow` to a `LinkExtractor`:
```python
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/sitemap.xml"]
sitemap_follow = LinkExtractor(allow=r"/posts-sitemap-\d+\.xml") # only post sitemaps
```
### Robots.txt support
Put a `robots.txt` URL directly in `sitemap_urls` and `SitemapSpider` will detect it, extract every sitemap shown, and follow each one:
```python
class MySitemap(SitemapSpider):
name = "sm"
sitemap_urls = ["https://example.com/robots.txt"]
```
### Alternate-language URLs
Set `sitemap_alternate_links = True` to also dispatch `<xhtml:link rel="alternate" hreflang="...">` URLs through your rules.
## Using `LinkExtractor` directly
You don't have to use the templates. `LinkExtractor` works inside any plain `Spider`:
```python
from scrapling.spiders import Spider, LinkExtractor
class CustomSpider(Spider):
name = "custom"
start_urls = ["https://example.com"]
def __init__(self):
super().__init__()
self._links = LinkExtractor(allow=r"/posts/", deny_domains="ads.example.com")
async def parse(self, response):
for url in self._links.extract(response):
yield response.follow(url, callback=self.parse_post)
async def parse_post(self, response):
yield {"title": response.css("h1::text").get()}
```
## LinkExtractor reference
| Argument | Default | Description |
|-------------------|----------------------|---------------------------------------------------------------------------------------------------|
| `allow` | `()` | URL patterns to keep. Empty means "match all". String, compiled `Pattern`, or iterable of either. |
| `deny` | `()` | URL patterns to drop. Always overrides `allow`. |
| `allow_domains` | `()` | Hostnames to keep. Subdomains match automatically (`example.com` matches `api.example.com`). |
| `deny_domains` | `()` | Hostnames to drop. |
| `restrict_css` | `()` | CSS selectors that scope DOM extraction to a region. |
| `restrict_xpath` | `()` | XPath selectors that scope DOM extraction to a region. |
| `tags` | `("a", "area")` | Element tags to look for links in. |
| `attrs` | `("href",)` | Attributes on those tags to read URLs from. |
| `canonicalize` | `True` | Sort query params and normalize the path. |
| `strip` | `True` | Strip whitespace from extracted URLs. |
| `keep_fragment` | `False` | Preserve the `#fragment` when canonicalizing. |
| `deny_extensions` | `IGNORED_EXTENSIONS` | File extensions to drop (pdf, zip, images, video, etc.). |
| `process` | `None` | Optional callable applied to each extracted URL before filtering. Return a falsy value to drop. |
`LinkExtractor.extract(response)` returns a `list[str]` of absolute, filtered, deduped URLs.
`LinkExtractor.matches(url)` returns a `bool` - the URL-only filter (allow/deny/domain/extension), used by `SitemapSpider` to dispatch URLs without a `Response`.
+5 -5
View File
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "scrapling"
# Static version instead of a dynamic version so we can get better layer caching while building docker, check the docker file to understand
version = "0.4.7"
version = "0.4.8"
description = "Scrapling is an undetectable, powerful, flexible, high-performance Python library that makes Web Scraping easy and effortless as it should be!"
readme = {file = "README.md", content-type = "text/markdown"}
license = {file = "LICENSE"}
@@ -61,7 +61,7 @@ classifiers = [
"Typing :: Typed",
]
dependencies = [
"lxml>=6.0.3",
"lxml>=6.1.0",
"cssselect>=1.4.0",
"orjson>=3.11.8",
"tld>=0.13.2",
@@ -73,10 +73,10 @@ dependencies = [
fetchers = [
"click>=8.3.0",
"curl_cffi>=0.15.0",
"playwright==1.58.0",
"patchright==1.58.2",
"playwright==1.59.0",
"patchright==1.59.1",
"browserforge>=1.2.4",
"apify-fingerprint-datapoints>=0.12.0",
"apify-fingerprint-datapoints>=0.13.0",
"msgspec>=0.21.1",
"anyio>=4.13.0",
"protego>=0.6.0",
+1 -1
View File
@@ -1,5 +1,5 @@
__author__ = "Karim Shoair (karim.shoair@pm.me)"
__version__ = "0.4.7"
__version__ = "0.4.8"
__copyright__ = "Copyright (c) 2024 Karim Shoair"
from typing import Any, TYPE_CHECKING
+2 -2
View File
@@ -13,8 +13,8 @@ from scrapling.core._types import Dict, Literal, Tuple
__OS_NAME__ = platform_system()
OSName = Literal["linux", "macos", "windows"]
# Current versions hardcoded for now (Playwright doesn't allow to know the version of a browser without launching it)
chromium_version = 145
chrome_version = 145
chromium_version = 147
chrome_version = 147
@lru_cache(1, typed=True)
+46 -9
View File
@@ -1,28 +1,65 @@
from scrapling.core._types import Any, Awaitable, Unpack
from scrapling.engines._browsers._types import DataRequestParams, GetRequestParams
from scrapling.engines.static import (
FetcherSession,
FetcherClient as _FetcherClient,
AsyncFetcherClient as _AsyncFetcherClient,
)
from scrapling.engines.toolbelt.custom import BaseFetcher
from scrapling.engines.toolbelt.custom import BaseFetcher, Response
__all__ = ["Fetcher", "AsyncFetcher", "FetcherSession"]
__FetcherClientInstance__ = _FetcherClient()
__AsyncFetcherClientInstance__ = _AsyncFetcherClient()
def _merge_selector_config(cls: type[BaseFetcher], kwargs: Any) -> Any:
"""Merge class-level parser arguments into per-request ``selector_config``.
Values from ``Fetcher.configure(...)`` act as the base; any explicit
``selector_config`` passed on the call overrides them.
"""
selector_config = kwargs.get("selector_config") or {}
kwargs["selector_config"] = {**cls._generate_parser_arguments(), **selector_config}
return kwargs
class Fetcher(BaseFetcher):
"""A basic `Fetcher` class type that can only do basic GET, POST, PUT, and DELETE HTTP requests based on `curl_cffi`."""
get = __FetcherClientInstance__.get
post = __FetcherClientInstance__.post
put = __FetcherClientInstance__.put
delete = __FetcherClientInstance__.delete
@classmethod
def get(cls, url: str, **kwargs: Unpack[GetRequestParams]) -> Response:
return __FetcherClientInstance__.get(url, **_merge_selector_config(cls, kwargs))
@classmethod
def post(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
return __FetcherClientInstance__.post(url, **_merge_selector_config(cls, kwargs))
@classmethod
def put(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
return __FetcherClientInstance__.put(url, **_merge_selector_config(cls, kwargs))
@classmethod
def delete(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
return __FetcherClientInstance__.delete(url, **_merge_selector_config(cls, kwargs))
class AsyncFetcher(BaseFetcher):
"""A basic `Fetcher` class type that can only do basic GET, POST, PUT, and DELETE HTTP requests based on `curl_cffi`."""
get = __AsyncFetcherClientInstance__.get
post = __AsyncFetcherClientInstance__.post
put = __AsyncFetcherClientInstance__.put
delete = __AsyncFetcherClientInstance__.delete
@classmethod
def get(cls, url: str, **kwargs: Unpack[GetRequestParams]) -> Awaitable[Response]:
return __AsyncFetcherClientInstance__.get(url, **_merge_selector_config(cls, kwargs))
@classmethod
def post(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
return __AsyncFetcherClientInstance__.post(url, **_merge_selector_config(cls, kwargs))
@classmethod
def put(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
return __AsyncFetcherClientInstance__.put(url, **_merge_selector_config(cls, kwargs))
@classmethod
def delete(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
return __AsyncFetcherClientInstance__.delete(url, **_merge_selector_config(cls, kwargs))
+9 -5
View File
@@ -519,7 +519,7 @@ class Selector(SelectorsGeneration):
def relocate(
self,
element: Union[Dict, HtmlElement, "Selector"],
percentage: int = 0,
percentage: int = 40,
selector_type: bool = False,
) -> Union[List[HtmlElement], "Selectors"]:
"""This function will search again for the element in the page tree, used automatically on page structure change
@@ -559,6 +559,10 @@ class Selector(SelectorsGeneration):
if not selector_type:
return score_table[highest_probability]
return self.__elements_convertor(score_table[highest_probability])
log.warning(
f"Adaptive relocation found no element above the {percentage}% threshold "
f"(top score: {highest_probability}%). Lower `percentage` if this is the right element."
)
return []
def css(
@@ -567,7 +571,7 @@ class Selector(SelectorsGeneration):
identifier: str = "",
adaptive: bool = False,
auto_save: bool = False,
percentage: int = 0,
percentage: int = 40,
) -> "Selectors":
"""Search the current tree with CSS3 selectors
@@ -627,7 +631,7 @@ class Selector(SelectorsGeneration):
identifier: str = "",
adaptive: bool = False,
auto_save: bool = False,
percentage: int = 0,
percentage: int = 40,
**kwargs: Any,
) -> "Selectors":
"""Search the current tree with XPath selectors
@@ -1220,7 +1224,7 @@ class Selectors(List[Selector]):
selector: str,
identifier: str = "",
auto_save: bool = False,
percentage: int = 0,
percentage: int = 40,
**kwargs: Any,
) -> "Selectors":
"""
@@ -1251,7 +1255,7 @@ class Selectors(List[Selector]):
selector: str,
identifier: str = "",
auto_save: bool = False,
percentage: int = 0,
percentage: int = 40,
) -> "Selectors":
"""
Call the ``.css()`` method for each element in this list and return
+6
View File
@@ -4,6 +4,8 @@ from .scheduler import Scheduler
from .engine import CrawlerEngine
from .session import SessionManager
from .spider import Spider, SessionConfigurationError
from .links import LinkExtractor
from .templates import CrawlSpider, SitemapSpider, CrawlRule
from scrapling.engines.toolbelt.custom import Response
__all__ = [
@@ -15,4 +17,8 @@ __all__ = [
"SessionManager",
"Scheduler",
"Response",
"LinkExtractor",
"CrawlSpider",
"CrawlRule",
"SitemapSpider",
]
+295
View File
@@ -0,0 +1,295 @@
"""Pure URL discovery primitive"""
import re
from urllib.parse import urlsplit
from w3lib.html import strip_html5_whitespace
from w3lib.url import canonicalize_url, safe_url_string
from scrapling.core._types import (
TYPE_CHECKING,
Iterable,
Callable,
List,
Optional,
Pattern,
Set,
Tuple,
Union,
Any,
)
from scrapling.core.utils import log
if TYPE_CHECKING:
from scrapling.engines.toolbelt.custom import Response
__all__ = ["LinkExtractor"]
valid_schemas = {"http", "https", "file"}
IGNORED_EXTENSIONS = {
# archives
"7z",
"7zip",
"bz2",
"rar",
"tar",
"tar.gz",
"xz",
"zip",
# images
"mng",
"pct",
"bmp",
"gif",
"jpg",
"jpeg",
"png",
"pst",
"psp",
"tif",
"tiff",
"ai",
"drw",
"dxf",
"eps",
"ps",
"svg",
"cdr",
"ico",
"webp",
# audio
"mp3",
"wma",
"ogg",
"wav",
"ra",
"aac",
"mid",
"au",
"aiff",
# video
"3gp",
"asf",
"asx",
"avi",
"mov",
"mp4",
"mpg",
"qt",
"rm",
"swf",
"wmv",
"m4a",
"m4v",
"flv",
"webm",
# office suites
"xls",
"xlsm",
"xlsx",
"xltm",
"xltx",
"potm",
"potx",
"ppt",
"pptm",
"pptx",
"pps",
"doc",
"docb",
"docm",
"docx",
"dotm",
"dotx",
"odt",
"ods",
"odg",
"odp",
# other
"css",
"pdf",
"exe",
"bin",
"rss",
"dmg",
"iso",
"apk",
"jar",
"sh",
"rb",
"js",
"hta",
"bat",
"cpl",
"msi",
"msp",
"py",
}
PatternInput = Iterable[Union[str, Pattern[str]]]
StrOrIterable = Union[str, Iterable[str]]
def _to_str_tuple(value: StrOrIterable) -> Tuple[str, ...]:
if not value:
return ()
if isinstance(value, str):
return (value,)
return tuple(value)
def _compile_patterns(patterns: Union[str, Pattern[str], PatternInput, None]) -> Tuple[Pattern[str], ...]:
if not patterns:
return ()
if isinstance(patterns, (str, re.Pattern)):
patterns = (patterns,)
return tuple(p if isinstance(p, re.Pattern) else re.compile(p) for p in patterns)
def _url_extension(url: str) -> str:
path = urlsplit(url).path
_, _, last = path.rpartition("/")
if "." not in last:
return ""
return last.rsplit(".", 1)[1].lower()
def _filler(x):
return x
class LinkExtractor:
"""Extracts and filters URLs from a `Response` (or a single URL via `matches`).
All matching is regex-based; allow/deny patterns can be plain strings (compiled
with `re.compile`) or pre-compiled `re.Pattern` objects, individually or as an
iterable.
:param allow: Regex pattern(s) URLs must match to be kept. String, compiled `re.Pattern`,
or an iterable of either. Empty means match all.
:param deny: Regex pattern(s) URLs must NOT match. Takes precedence over `allow`.
:param allow_domains: Domain(s) to keep. Matches the exact host or any subdomain
(e.g. `"example.com"` matches `"api.example.com"`). String or iterable.
:param deny_domains: Domain(s) to exclude. Same matching rules as `allow_domains`.
:param restrict_css: CSS selectors to scope DOM extraction to. Empty means whole page.
:param restrict_xpath: XPath selectors to scope DOM extraction to. Empty means whole page.
:param tags: Element tags to look for links in. Default ("a", "area").
:param attrs: Attributes on those tags to read URLs from. Default ("href",).
:param canonicalize: Canonicalize URLs (sort query params, normalize path). Default True.
:param strip: Strip whitespace from extracted URLs. Default True.
:param keep_fragment: Preserve the URL fragment when canonicalizing. Default False.
:param deny_extensions: File extensions to drop. Default `IGNORED_EXTENSIONS`.
:param process: A function to do a process on the values extracted before using them. Return None to drop any value.
"""
def __init__(
self,
allow: Union[str, Pattern[str], PatternInput] = (),
deny: Union[str, Pattern[str], PatternInput] = (),
allow_domains: StrOrIterable = (),
deny_domains: StrOrIterable = (),
restrict_css: StrOrIterable = (),
restrict_xpath: StrOrIterable = (),
tags: Iterable[str] = ("a", "area"),
attrs: Iterable[str] = ("href",),
canonicalize: bool = True,
strip: bool = True,
keep_fragment: bool = False,
deny_extensions: Optional[Iterable[str]] = None,
process: Callable[[Any], Any] | None = None,
) -> None:
self.allow: Tuple[Pattern[str], ...] = _compile_patterns(allow)
self.deny: Tuple[Pattern[str], ...] = _compile_patterns(deny)
self.allow_domains: Tuple[str, ...] = tuple(d.lower() for d in _to_str_tuple(allow_domains))
self.deny_domains: Tuple[str, ...] = tuple(d.lower() for d in _to_str_tuple(deny_domains))
self.restrict_css: Tuple[str, ...] = _to_str_tuple(restrict_css)
self.restrict_xpath: Tuple[str, ...] = _to_str_tuple(restrict_xpath)
self.tags: Tuple[str, ...] = tuple(tags)
self.attrs: Tuple[str, ...] = tuple(attrs)
self.canonicalize = canonicalize
self.strip = strip
self.keep_fragment = keep_fragment
self.deny_extensions: Set[str] = set(
(ext.lower().lstrip(".") for ext in deny_extensions) if deny_extensions is not None else IGNORED_EXTENSIONS
)
self.process: Callable[[Any], Any] = process if callable(process) else _filler
def extract(self, response: "Response") -> List[str]:
"""Return absolute, filtered, deduped URLs from `response`."""
scopes: List[Any] = []
if self.restrict_xpath:
for xp in self.restrict_xpath:
scopes.extend(response.xpath(xp))
if self.restrict_css:
for cs in self.restrict_css:
scopes.extend(response.css(cs))
if not scopes:
scopes = [response]
out: List[str] = []
search_selector = "| ".join([f".//{tag}/@{attr}" for tag in self.tags for attr in self.attrs])
for scope in scopes:
for url in scope._root.xpath(search_selector):
if not url:
continue
url = str(url)
if self.strip:
url = strip_html5_whitespace(url)
if not url:
continue
url = str(response.urljoin(url))
url = self.process(url)
if not url:
continue
if self.canonicalize:
url = canonicalize_url(url, keep_fragments=self.keep_fragment)
try:
url = safe_url_string(url, encoding=response.encoding)
except ValueError:
log.debug(f"Skipping the extraction of bad URL {url!r}")
continue
if not self._url_passes(url):
continue
out.append(url)
# Switching to dict for deduplication instead of Set will keep the insertion order of the links.
return list(dict.fromkeys(out))
def matches(self, url: str) -> bool:
"""URL-only filter (no response extraction).
Applies allow/deny/allow_domains/deny_domains/deny_extensions to a single URL.
Used by `SitemapSpider` to dispatch sitemap URLs through `CrawlRule`s without
needing a `Response`.
"""
if self.canonicalize:
url = canonicalize_url(url, keep_fragments=self.keep_fragment)
return self._url_passes(url)
def _url_passes(self, url: str) -> bool:
if url.split("://", 1)[0] not in valid_schemas:
return False
ext = _url_extension(url)
if ext and ext in self.deny_extensions:
return False
if self.allow and not any(p.search(url) for p in self.allow):
return False
if self.deny and any(p.search(url) for p in self.deny):
return False
if self.allow_domains or self.deny_domains:
host = (urlsplit(url).hostname or "").lower()
if self.allow_domains and not any(host == d or host.endswith("." + d) for d in self.allow_domains):
return False
if self.deny_domains and any(host == d or host.endswith("." + d) for d in self.deny_domains):
return False
return True
+14 -3
View File
@@ -22,6 +22,13 @@ def _convert_to_bytes(value: str | bytes) -> bytes:
return value.encode(encoding="utf-8", errors="ignore")
def _stable_value_repr(value: Any) -> str:
try:
return orjson.dumps(value, option=orjson.OPT_SORT_KEYS, default=repr).decode()
except TypeError:
return repr(value)
class Request:
def __init__(
self,
@@ -97,15 +104,19 @@ class Request:
}
if include_kwargs:
kwargs = (key.lower() for key in self._session_kwargs.keys() if key.lower() not in ("data", "json"))
data["kwargs"] = "".join(set(_convert_to_bytes(key).hex() for key in kwargs))
filtered_kwargs = {
key.lower(): _stable_value_repr(value)
for key, value in self._session_kwargs.items()
if key.lower() not in ("data", "json")
}
data["kwargs"] = tuple(sorted(filtered_kwargs.items()))
if include_headers:
headers = self._session_kwargs.get("headers") or self._session_kwargs.get("extra_headers") or {}
processed_headers = {}
# Some header normalization
for key, value in headers.items():
processed_headers[_convert_to_bytes(key.lower()).hex()] = _convert_to_bytes(value.lower()).hex()
processed_headers[_convert_to_bytes(key.lower()).hex()] = _convert_to_bytes(value).hex()
data["headers"] = tuple(processed_headers.items())
fp = hashlib.sha1(orjson.dumps(data, option=orjson.OPT_SORT_KEYS), usedforsecurity=False).digest()
+8
View File
@@ -0,0 +1,8 @@
from .crawler import CrawlSpider, CrawlRule
from .sitemap import SitemapSpider
__all__ = [
"CrawlSpider",
"CrawlRule",
"SitemapSpider",
]
+72
View File
@@ -0,0 +1,72 @@
"""Generic spider templates that build on the `Spider` base."""
from dataclasses import dataclass
from scrapling.spiders.links import LinkExtractor
from scrapling.spiders.request import Request
from scrapling.spiders.spider import Spider
from scrapling.core._types import (
TYPE_CHECKING,
Any,
AsyncGenerator,
Callable,
Dict,
List,
Optional,
Union,
)
if TYPE_CHECKING:
from scrapling.engines.toolbelt.custom import Response
__all__ = ["CrawlRule", "CrawlSpider"]
ParseCallback = Callable[
["Response"],
AsyncGenerator[Union[Dict[str, Any], Request, None], None],
]
ProcessRequestFn = Callable[[Request, "Response"], Request]
@dataclass
class CrawlRule:
"""Rule for `CrawlSpider`: extract links from a response and dispatch them.
:param link_extractor: `LinkExtractor` that produces URLs from each response.
:param callback: Bound method on the spider to call for each matched URL.
Falls back to the spider's default ``parse()`` by default.
:param priority: Override the priority of the requests that will be dispatched.
:param process_request: Optional bound method to mutate each `Request` before
it is yielded. Signature: ``(request, response) -> request``. Use it to
add headers, change priority, or filter requests.
"""
link_extractor: LinkExtractor
callback: Optional[ParseCallback] = None
priority: Optional[int] = None
process_request: Optional[ProcessRequestFn] = None
class CrawlSpider(Spider):
"""A generic spider that can extract and follow links automatically based on crawl rules.
Override `rules()` to return a list of `CrawlRule`s.
You can start from it and override it as needed for more custom functionality, or just implement your own spider.
"""
def rules(self) -> List[CrawlRule]:
"""Override to define link-following rules."""
return []
async def parse(self, response: "Response") -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
for rule in self.rules():
for url in rule.link_extractor.extract(response):
req = response.follow(url, callback=rule.callback)
if rule.priority is not None:
req.priority = rule.priority
if rule.process_request is not None:
req = rule.process_request(req, response)
yield req
+193
View File
@@ -0,0 +1,193 @@
"""Sitemap template spider."""
from dataclasses import dataclass, field
from gzip import GzipFile
from io import BytesIO
from urllib.parse import urlsplit
from lxml import etree
from protego import Protego
from scrapling.core._types import (
TYPE_CHECKING,
Any,
AsyncGenerator,
Dict,
List,
Optional,
Union,
)
from scrapling.spiders.links import LinkExtractor
from scrapling.spiders.request import Request
from scrapling.spiders.spider import Spider
from scrapling.spiders.templates.crawler import CrawlRule
if TYPE_CHECKING:
from scrapling.engines.toolbelt.custom import Response
__all__ = ["SitemapSpider"]
_GZIP_MAGIC = b"\x1f\x8b"
_GUNZIP_MAX_SIZE = 64 * 1024 * 1024 # 64 MiB cap, defends against gzip bombs
@dataclass
class SitemapResult:
"""Parsed sitemap body.
`urls` holds the entries from a `<urlset>`; `sitemaps` holds child sitemap
URLs from a `<sitemapindex>` (each of which is fetched recursively).
"""
urls: List[str] = field(default_factory=list)
sitemaps: List[str] = field(default_factory=list)
class SitemapSpider(Spider):
"""A Spider that seeds a crawl from sitemap(s), and follows the rules.
Override `rules()` to return a list of `CrawlRule`s.
If there are no rules provided, all non-sitemap urls will be redirected to `parse()`, which must be overridden or it will raise `NotImplementedError`.
:cvar sitemap_urls: Explicit list of sitemap (or robots.txt) URLs to fetch.
:cvar sitemap_follow: `LinkExtractor` filtering which child sitemaps inside a
`<sitemapindex>` to descend into. ``None`` means descend into all.
:cvar sitemap_alternate_links: When enabled, alternate-language URLs are also
routed through `rules()`.
"""
sitemap_urls: List[str] = []
sitemap_follow: Optional[LinkExtractor] = None
sitemap_alternate_links: bool = False
def rules(self) -> List[CrawlRule]:
"""Override to define dispatch rules for sitemap URLs."""
return []
async def start_requests(self) -> AsyncGenerator[Request, None]:
if self.sitemap_urls:
for url in self.sitemap_urls:
yield Request(url, callback=self._parse_sitemap)
return
raise RuntimeError("`SitemapSpider` needs `sitemap_urls` to be set.")
async def parse(self, response: "Response") -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
"""Default callback for processing responses"""
raise NotImplementedError(f"{self.__class__.__name__} must implement parse() method")
yield # Make this a generator for type checkers
def _robots_body(self, response: "Response") -> List[str]:
"""Extract `Sitemap` directives from a robots.txt body via protego."""
try:
text = response.body.decode(response.encoding, errors="replace")
parser = Protego.parse(text)
except Exception as e:
self.logger.warning(f"Failed to parse robots.txt: {e}")
return []
return list(parser.sitemaps)
@staticmethod
def _decompress(body: bytes, content_type: Optional[str]) -> bytes:
if (content_type and ("gzip" in content_type.lower())) or (body[:2] == _GZIP_MAGIC):
out = bytearray()
with GzipFile(fileobj=BytesIO(body)) as f:
while chunk := f.read1(8192):
out.extend(chunk)
if len(out) > _GUNZIP_MAX_SIZE:
raise OSError(f"gzip output exceeds {_GUNZIP_MAX_SIZE} bytes")
return bytes(out)
return body
def _extract_urls(self, root: Any) -> List[str]:
urls: List[str] = []
for url_el in root:
if self._get_type(url_el) != "url":
continue
for child in url_el:
name = self._get_type(child)
if name == "loc" and child.text:
urls.append(child.text.strip())
elif self.sitemap_alternate_links and name == "link":
href = child.get("href")
if href:
urls.append(href.strip())
return urls
@staticmethod
def _get_type(el: Any) -> str:
return etree.QName(el.tag).localname
def _sm_body(self, body: bytes, content_type: Optional[str] = None) -> SitemapResult:
"""Parse a sitemap body and return its URLs and any child sitemaps."""
try:
body = self._decompress(body, content_type)
except OSError as e:
self.logger.warning(f"Failed to decompress sitemap: {e}")
return SitemapResult()
try:
root = etree.fromstring(body)
except etree.XMLSyntaxError as e:
self.logger.warning(f"Failed to parse sitemap XML: {e}")
return SitemapResult()
root_name = self._get_type(root)
if root_name == "sitemapindex":
locs = []
for sm_el in root:
if self._get_type(sm_el) == "sitemap":
for child in sm_el:
if self._get_type(child) == "loc" and child.text:
locs.append(child.text.strip())
break
return SitemapResult(sitemaps=locs)
if root_name == "urlset":
return SitemapResult(urls=self._extract_urls(root))
self.logger.warning(f"Unknown sitemap root element: {root_name!r}")
return SitemapResult()
async def _parse_sitemap(self, response: "Response") -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
if urlsplit(response.url).path.endswith("/robots.txt"):
sitemaps = self._robots_body(response)
if not sitemaps:
self.logger.warning(f"No Sitemaps found in {response.url}")
for sitemap_url in sitemaps:
yield response.follow(sitemap_url, callback=self._parse_sitemap)
return
content_type = response.headers.get("content-type") if response.headers else None
result = self._sm_body(response.body, content_type=content_type)
# Descend into child sitemaps (apply sitemap_follow filter if present)
for child_url in result.sitemaps:
if self.sitemap_follow is not None and not self.sitemap_follow.matches(child_url):
continue
yield response.follow(child_url, callback=self._parse_sitemap)
# Dispatch each URL through rules() (first match wins; unmatched drop unless rules empty)
rules = self.rules()
for url in result.urls:
req = self._dispatch(response, url, rules)
if req is not None:
yield req
@staticmethod
def _dispatch(response: "Response", url: str, rules: List[CrawlRule]) -> Optional[Request]:
if not rules:
return response.follow(url)
for rule in rules:
if rule.link_extractor.matches(url):
req = response.follow(url, callback=rule.callback)
if rule.priority is not None:
req.priority = rule.priority
if rule.process_request is not None:
req = rule.process_request(req, response)
return req
return None
+2 -2
View File
@@ -14,12 +14,12 @@
"mimeType": "image/png"
}
],
"version": "0.4.7",
"version": "0.4.8",
"packages": [
{
"registryType": "pypi",
"identifier": "scrapling",
"version": "0.4.7",
"version": "0.4.8",
"runtimeHint": "uvx",
"packageArguments": [
{
+1 -1
View File
@@ -1,6 +1,6 @@
[metadata]
name = scrapling
version = 0.4.7
version = 0.4.8
author = Karim Shoair
author_email = karim.shoair@pm.me
description = Scrapling is an undetectable, powerful, flexible, high-performance Python library that makes Web Scraping easy and effortless as it should be!
+36
View File
@@ -6,6 +6,17 @@ from scrapling.fetchers import AsyncFetcher
AsyncFetcher.adaptive = True
@pytest.fixture
def _reset_async_fetcher_config():
"""Snapshot and restore the mutable class-level parser config around a test."""
snapshot = {k: getattr(AsyncFetcher, k) for k in AsyncFetcher.parser_keywords}
try:
yield
finally:
for k, v in snapshot.items():
setattr(AsyncFetcher, k, v)
@pytest_httpbin.use_class_based_httpbin
@pytest.mark.asyncio
class TestAsyncFetcher:
@@ -124,3 +135,28 @@ class TestAsyncFetcher:
timeout=None,
)
).status == 200
async def test_configure_propagates_to_response(
self, fetcher, urls, _reset_async_fetcher_config
):
"""`AsyncFetcher.configure()` must reach the Response's Selector on the HTTP path."""
AsyncFetcher.configure(adaptive=False, adaptive_domain="")
baseline = await fetcher.get(urls["html_url"])
assert baseline._storage is None
AsyncFetcher.configure(adaptive=True, adaptive_domain="configured.test")
configured = await fetcher.get(urls["html_url"])
assert configured._storage is not None
assert configured.url == "configured.test"
async def test_selector_config_overrides_configure(
self, fetcher, urls, _reset_async_fetcher_config
):
"""A per-request ``selector_config`` overrides the class-level configure()."""
AsyncFetcher.configure(adaptive=True, adaptive_domain="from-configure.test")
response = await fetcher.get(
urls["html_url"],
selector_config={"adaptive_domain": "from-request.test"},
)
assert response._storage is not None
assert response.url == "from-request.test"
+32
View File
@@ -6,6 +6,17 @@ from scrapling import Fetcher
Fetcher.adaptive = True
@pytest.fixture
def _reset_fetcher_config():
"""Snapshot and restore the mutable class-level parser config around a test."""
snapshot = {k: getattr(Fetcher, k) for k in Fetcher.parser_keywords}
try:
yield
finally:
for k, v in snapshot.items():
setattr(Fetcher, k, v)
@pytest_httpbin.use_class_based_httpbin
class TestFetcher:
@pytest.fixture(scope="class")
@@ -119,3 +130,24 @@ class TestFetcher:
).status
== 200
)
def test_configure_propagates_to_response(self, fetcher, _reset_fetcher_config):
"""`Fetcher.configure()` must reach the Response's Selector on the HTTP path."""
Fetcher.configure(adaptive=False, adaptive_domain="")
baseline = fetcher.get(self.html_url)
assert baseline._storage is None
Fetcher.configure(adaptive=True, adaptive_domain="configured.test")
configured = fetcher.get(self.html_url)
assert configured._storage is not None
assert configured.url == "configured.test"
def test_selector_config_overrides_configure(self, fetcher, _reset_fetcher_config):
"""A per-request ``selector_config`` overrides the class-level configure()."""
Fetcher.configure(adaptive=True, adaptive_domain="from-configure.test")
response = fetcher.get(
self.html_url,
selector_config={"adaptive_domain": "from-request.test"},
)
assert response._storage is not None
assert response.url == "from-request.test"
+1 -1
View File
@@ -1,6 +1,6 @@
pytest>=2.8.0,<9
pytest-cov
playwright==1.58.0
playwright==1.59.0
werkzeug<3.0.0
pytest-httpbin==2.1.0
pytest-asyncio
+243
View File
@@ -0,0 +1,243 @@
"""Tests for `LinkExtractor`."""
import re
import pytest
from scrapling.engines.toolbelt.custom import Response
from scrapling.spiders.links import IGNORED_EXTENSIONS, LinkExtractor
def _make_response(html: str, url: str = "https://example.com/page") -> Response:
"""Build a minimal Response wrapping the given HTML."""
return Response(
url=url,
content=html,
status=200,
reason="OK",
cookies={},
headers={},
request_headers={},
)
HTML_BASIC = """
<html><body>
<a href="/posts/1">post 1</a>
<a href="/posts/2">post 2</a>
<a href="https://other.com/page">external</a>
<a href="/about">about</a>
<a href="mailto:x@example.com">mail</a>
<a href="javascript:alert(1)">js</a>
<a href="/file.pdf">pdf</a>
<area href="/area-link">area</area>
<link rel="stylesheet" href="/style.css">
</body></html>
"""
class TestExtractBasic:
def test_default_extracts_a_and_area_with_href(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor().extract(resp)
# mailto/javascript filtered (non-http scheme), .pdf filtered (deny_extensions)
# link[rel=stylesheet] not in default tags
assert "https://example.com/posts/1" in urls
assert "https://example.com/posts/2" in urls
assert "https://example.com/about" in urls
assert "https://example.com/area-link" in urls
assert "https://other.com/page" in urls
assert all(not u.startswith("mailto:") for u in urls)
assert all(not u.startswith("javascript:") for u in urls)
assert not any(u.endswith(".pdf") for u in urls)
assert not any(u.endswith(".css") for u in urls)
def test_relative_urls_become_absolute_via_urljoin(self):
resp = _make_response('<a href="foo/bar">x</a>', url="https://example.com/sub/")
assert LinkExtractor().extract(resp) == ["https://example.com/sub/foo/bar"]
def test_empty_allow_means_match_all(self):
resp = _make_response('<a href="/anything">x</a><a href="/else">y</a>')
out = LinkExtractor().extract(resp)
assert len(out) == 2
class TestAllowDeny:
def test_allow_regex_filters_in(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(allow=r"/posts/").extract(resp)
assert urls == ["https://example.com/posts/1", "https://example.com/posts/2"]
def test_deny_regex_filters_out(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(deny=r"/posts/").extract(resp)
assert "https://example.com/posts/1" not in urls
assert "https://example.com/about" in urls
def test_deny_overrides_allow(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(allow=r"/posts/", deny=r"/posts/2").extract(resp)
assert urls == ["https://example.com/posts/1"]
def test_compiled_pattern_accepted(self):
resp = _make_response(HTML_BASIC)
pat = re.compile(r"/posts/\d+$")
urls = LinkExtractor(allow=pat).extract(resp)
assert len(urls) == 2
def test_iterable_of_patterns(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(allow=[r"/posts/", r"/about"]).extract(resp)
assert "https://example.com/posts/1" in urls
assert "https://example.com/about" in urls
class TestDomains:
def test_allow_domains_keeps_only_matching(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(allow_domains="example.com").extract(resp)
assert all("example.com" in u for u in urls)
assert "https://other.com/page" not in urls
def test_allow_domains_matches_subdomains(self):
html = '<a href="https://api.example.com/x">a</a><a href="https://other.com/y">b</a>'
resp = _make_response(html)
urls = LinkExtractor(allow_domains="example.com").extract(resp)
assert urls == ["https://api.example.com/x"]
def test_deny_domains_filters_out(self):
resp = _make_response(HTML_BASIC)
urls = LinkExtractor(deny_domains="other.com").extract(resp)
assert "https://other.com/page" not in urls
assert "https://example.com/posts/1" in urls
class TestRestrict:
def test_restrict_css_scopes_extraction(self):
html = """
<html><body>
<nav><a href="/nav-link">n</a></nav>
<main><a href="/main-link">m</a></main>
</body></html>
"""
resp = _make_response(html)
urls = LinkExtractor(restrict_css="main").extract(resp)
assert urls == ["https://example.com/main-link"]
def test_restrict_xpath_scopes_extraction(self):
html = """
<html><body>
<div id="header"><a href="/h">h</a></div>
<div id="content"><a href="/c">c</a></div>
</body></html>
"""
resp = _make_response(html)
urls = LinkExtractor(restrict_xpath='//div[@id="content"]').extract(resp)
assert urls == ["https://example.com/c"]
class TestTagsAttrs:
def test_custom_tags_and_attrs_for_stylesheets(self):
html = '<link rel="stylesheet" href="/style.css"><a href="/page">p</a>'
resp = _make_response(html)
# Override deny_extensions to allow .css through, and pick up <link href>
urls = LinkExtractor(tags=("link",), attrs=("href",), deny_extensions=()).extract(resp)
assert urls == ["https://example.com/style.css"]
class TestCanonicalization:
def test_query_params_sorted(self):
resp = _make_response('<a href="/x?b=2&a=1">x</a>')
urls = LinkExtractor().extract(resp)
assert urls == ["https://example.com/x?a=1&b=2"]
def test_fragment_dropped_by_default(self):
resp = _make_response('<a href="/x#section">x</a>')
urls = LinkExtractor().extract(resp)
assert urls == ["https://example.com/x"]
def test_keep_fragment_preserves_it(self):
resp = _make_response('<a href="/x#section">x</a>')
urls = LinkExtractor(keep_fragment=True).extract(resp)
assert urls == ["https://example.com/x#section"]
def test_canonicalize_off_leaves_url_unchanged(self):
resp = _make_response('<a href="/x?b=2&a=1#f">x</a>')
urls = LinkExtractor(canonicalize=False).extract(resp)
assert urls == ["https://example.com/x?b=2&a=1#f"]
class TestDedup:
def test_unique_drops_duplicates(self):
html = '<a href="/x">a</a><a href="/x">b</a><a href="/x?">c</a>'
resp = _make_response(html)
urls = LinkExtractor().extract(resp)
# canonicalize collapses /x and /x? together
assert urls == ["https://example.com/x"]
class TestExtensions:
def test_default_deny_extensions_drops_pdf_zip_images(self):
html = '<a href="/a.pdf">pdf</a><a href="/b.zip">zip</a><a href="/c.png">png</a><a href="/d">ok</a>'
resp = _make_response(html)
urls = LinkExtractor().extract(resp)
assert urls == ["https://example.com/d"]
def test_custom_deny_extensions_overrides_default(self):
html = '<a href="/a.pdf">pdf</a><a href="/b.zip">zip</a>'
resp = _make_response(html)
urls = LinkExtractor(deny_extensions={"zip"}).extract(resp)
# .pdf now allowed because we replaced the set
assert urls == ["https://example.com/a.pdf"]
def test_empty_deny_extensions_allows_everything(self):
html = '<a href="/a.pdf">pdf</a>'
resp = _make_response(html)
urls = LinkExtractor(deny_extensions=()).extract(resp)
assert urls == ["https://example.com/a.pdf"]
class TestStrip:
def test_strip_removes_whitespace(self):
resp = _make_response('<a href=" /spaced ">x</a>')
urls = LinkExtractor().extract(resp)
assert urls == ["https://example.com/spaced"]
class TestMatches:
def test_matches_honors_allow(self):
ex = LinkExtractor(allow=r"/posts/")
assert ex.matches("https://example.com/posts/1") is True
assert ex.matches("https://example.com/about") is False
def test_matches_honors_deny(self):
ex = LinkExtractor(deny=r"/admin")
assert ex.matches("https://example.com/admin/x") is False
assert ex.matches("https://example.com/posts/1") is True
def test_matches_honors_allow_domains(self):
ex = LinkExtractor(allow_domains="example.com")
assert ex.matches("https://api.example.com/x") is True
assert ex.matches("https://other.com/x") is False
def test_matches_honors_deny_extensions(self):
ex = LinkExtractor()
assert ex.matches("https://example.com/file.pdf") is False
assert ex.matches("https://example.com/page") is True
def test_matches_rejects_non_http_schemes(self):
ex = LinkExtractor()
assert ex.matches("mailto:x@example.com") is False
assert ex.matches("javascript:void(0)") is False
assert ex.matches("ftp://example.com/x") is False
def test_matches_canonicalizes_before_checking(self):
ex = LinkExtractor(allow=r"a=1&b=2$")
# the URL has params in the wrong order; canonicalize sorts them
assert ex.matches("https://example.com/x?b=2&a=1") is True
class TestIgnoredExtensions:
def test_constant_includes_common_binary_types(self):
for ext in ("pdf", "zip", "png", "mp4", "exe"):
assert ext in IGNORED_EXTENSIONS
+32
View File
@@ -99,6 +99,38 @@ class TestRequestProperties:
r2 = Request("https://example.com/page2")
assert r1.update_fingerprint() != r2.update_fingerprint()
def test_fingerprint_include_kwargs_uses_kwarg_values(self):
"""Test kwargs with different values produce different fingerprints."""
r1 = Request("https://example.com", timeout=1)
r2 = Request("https://example.com", timeout=2)
assert r1.update_fingerprint(include_kwargs=True) != r2.update_fingerprint(include_kwargs=True)
def test_fingerprint_include_kwargs_handles_non_primitive_values(self):
class _Opaque:
def __repr__(self) -> str:
return "_Opaque(stable)"
opaque = _Opaque()
r1 = Request("https://example.com", proxies={"http": "p1"}, custom=opaque)
r2 = Request("https://example.com", proxies={"http": "p1"}, custom=opaque)
r3 = Request("https://example.com", proxies={"http": "p2"}, custom=opaque)
fp1 = r1.update_fingerprint(include_kwargs=True)
r2._fp = None
fp2 = r2.update_fingerprint(include_kwargs=True)
fp3 = r3.update_fingerprint(include_kwargs=True)
assert fp1 == fp2
assert fp1 != fp3
def test_fingerprint_include_headers_preserves_header_value_case(self):
"""Test header values are fingerprinted without lowercasing."""
r1 = Request("https://example.com", headers={"X-Test": "A"})
r2 = Request("https://example.com", headers={"X-Test": "a"})
assert r1.update_fingerprint(include_headers=True) != r2.update_fingerprint(include_headers=True)
class TestRequestCopy:
"""Test Request copy functionality."""
+245
View File
@@ -0,0 +1,245 @@
"""Tests for `SitemapSpider`."""
import gzip
import pickle
import pytest
from scrapling.engines.toolbelt.custom import Response
from scrapling.spiders.links import LinkExtractor
from scrapling.spiders.request import Request
from scrapling.spiders.templates.sitemap import SitemapSpider
from scrapling.spiders.templates import CrawlRule
from scrapling.core._types import AsyncGenerator
URLSET_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
<url>
<loc>https://example.com/posts/1</loc>
<lastmod>2026-01-15</lastmod>
<changefreq>daily</changefreq>
<priority>0.8</priority>
</url>
<url>
<loc>https://example.com/posts/2</loc>
<lastmod>2026-02-20</lastmod>
</url>
<url>
<loc>https://example.com/about</loc>
</url>
</urlset>
"""
URLSET_WITH_ALTERNATES = b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
xmlns:xhtml="http://www.w3.org/1999/xhtml">
<url>
<loc>https://example.com/en/page</loc>
<xhtml:link rel="alternate" hreflang="fr" href="https://example.com/fr/page"/>
<xhtml:link rel="alternate" hreflang="de" href="https://example.com/de/page"/>
</url>
</urlset>
"""
INDEX_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
<sitemap><loc>https://example.com/posts-sitemap.xml</loc></sitemap>
<sitemap><loc>https://example.com/products-sitemap.xml</loc></sitemap>
<sitemap><loc>https://example.com/skip-sitemap.xml</loc></sitemap>
</sitemapindex>
"""
def _make_response(body: bytes, url: str = "https://example.com/sitemap.xml", headers: dict | None = None) -> Response:
resp = Response(
url=url,
content=body,
status=200,
reason="OK",
cookies={},
headers=headers or {},
request_headers={},
)
resp.request = Request(url, sid="default")
return resp
async def _collect(agen: AsyncGenerator) -> list:
return [item async for item in agen]
class TestSitemapSpiderFlow:
@pytest.mark.asyncio
async def test_urlset_dispatched_through_rules(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
async def parse_post(self, response):
yield {"post": response.url}
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
post_reqs = [r for r in out if "/posts/" in r.url]
about_reqs = [r for r in out if "/about" in r.url]
# Two posts dispatched to parse_post; /about is dropped (matches no rule, non-empty rules)
assert len(post_reqs) == 2
assert all(r.callback == spider.parse_post for r in post_reqs)
assert about_reqs == []
@pytest.mark.asyncio
async def test_no_rules_means_all_urls_fall_through(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
assert len(out) == 3
assert all(r.callback is None for r in out)
@pytest.mark.asyncio
async def test_sitemapindex_descends_into_children(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(INDEX_XML)))
# No urls in this index, just three child sitemap fetches
assert len(out) == 3
assert all(r.callback == spider._parse_sitemap for r in out)
assert {r.url for r in out} == {
"https://example.com/posts-sitemap.xml",
"https://example.com/products-sitemap.xml",
"https://example.com/skip-sitemap.xml",
}
@pytest.mark.asyncio
async def test_sitemap_follow_filters_child_sitemaps(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
sitemap_follow = LinkExtractor(allow=r"posts-sitemap")
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(INDEX_XML)))
assert {r.url for r in out} == {"https://example.com/posts-sitemap.xml"}
@pytest.mark.asyncio
async def test_alternate_links_dispatched_when_enabled(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
sitemap_alternate_links = True
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(URLSET_WITH_ALTERNATES)))
urls = {r.url for r in out}
assert urls == {
"https://example.com/en/page",
"https://example.com/fr/page",
"https://example.com/de/page",
}
@pytest.mark.asyncio
async def test_gzipped_sitemap_handled_via_magic_bytes(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml.gz"]
spider = S()
body = gzip.compress(URLSET_XML)
out = await _collect(spider._parse_sitemap(_make_response(body, url="https://example.com/sitemap.xml.gz")))
assert len(out) == 3
class TestSitemapSpiderStartRequests:
@pytest.mark.asyncio
async def test_start_requests_uses_sitemap_urls(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://a.com/s.xml", "https://b.com/s.xml"]
spider = S()
out = [req async for req in spider.start_requests()]
assert {r.url for r in out} == {"https://a.com/s.xml", "https://b.com/s.xml"}
assert all(r.callback == spider._parse_sitemap for r in out)
@pytest.mark.asyncio
async def test_start_requests_raises_when_nothing_configured(self):
class S(SitemapSpider):
name = "s"
spider = S()
with pytest.raises(RuntimeError, match="needs `sitemap_urls`"):
[req async for req in spider.start_requests()]
class TestRobotsTxt:
@pytest.mark.asyncio
async def test_parse_sitemap_yields_requests_from_robots_directives(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/robots.txt"]
spider = S()
body = b"User-agent: *\nSitemap: https://example.com/sitemap.xml\n"
resp = _make_response(body, url="https://example.com/robots.txt")
out = await _collect(spider._parse_sitemap(resp))
assert len(out) == 1
assert out[0].url == "https://example.com/sitemap.xml"
assert out[0].callback == spider._parse_sitemap
@pytest.mark.asyncio
async def test_parse_sitemap_robots_with_no_directives_warns(self):
# Spider's logger has propagate=False, so we attach our own handler to it.
import logging
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/robots.txt"]
spider = S()
records: list[logging.LogRecord] = []
class _Capture(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
records.append(record)
spider.logger.addHandler(_Capture())
body = b"User-agent: *\nDisallow: /\n"
resp = _make_response(body, url="https://example.com/robots.txt")
out = await _collect(spider._parse_sitemap(resp))
assert out == []
assert any("No Sitemaps" in r.getMessage() for r in records if r.levelno == logging.WARNING)
class TestSitemapSpiderPickle:
@pytest.mark.asyncio
async def test_pickle_request_with_bound_method_callback_via_rules(self):
class S(SitemapSpider):
name = "s"
sitemap_urls = ["https://example.com/sitemap.xml"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
async def parse_post(self, response):
yield {"post": response.url}
spider = S()
out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
post_req = next(r for r in out if "/posts/" in r.url)
state = post_req.__getstate__()
assert state["_callback_name"] == "parse_post"
# Round-trip
pickled = pickle.dumps(post_req)
restored = pickle.loads(pickled)
fresh = S()
restored._restore_callback(fresh)
assert restored.callback == fresh.parse_post
+230
View File
@@ -0,0 +1,230 @@
"""Tests for `CrawlSpider` and `CrawlRule`."""
import pickle
import pytest
from scrapling.engines.toolbelt.custom import Response
from scrapling.spiders.links import LinkExtractor
from scrapling.spiders.request import Request
from scrapling.spiders.templates import CrawlRule, CrawlSpider
from scrapling.core._types import Any, AsyncGenerator, Dict, Union
HTML = """
<html><body>
<a href="/posts/1">post 1</a>
<a href="/posts/2">post 2</a>
<a href="/page/2/">next page</a>
<a href="/about">about</a>
</body></html>
"""
def _make_response(url: str = "https://example.com/") -> Response:
"""Build a Response with a Request attached so `response.follow()` works."""
resp = Response(
url=url,
content=HTML,
status=200,
reason="OK",
cookies={},
headers={},
request_headers={},
)
resp.request = Request(url)
return resp
class _TestSpider(CrawlSpider):
name = "test"
start_urls = ["https://example.com/"]
async def parse_post(self, response: Response) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
yield {"post": response.url}
async def parse_page(self, response: Response) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
yield {"page": response.url}
async def _collect(agen: AsyncGenerator) -> list:
return [item async for item in agen]
class TestCrawlSpider:
@pytest.mark.asyncio
async def test_empty_rules_yields_nothing(self):
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
spider = S()
out = await _collect(spider.parse(_make_response()))
assert out == []
@pytest.mark.asyncio
async def test_single_rule_yields_matching_links(self):
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
spider = S()
out = await _collect(spider.parse(_make_response()))
urls = [r.url for r in out]
assert urls == ["https://example.com/posts/1", "https://example.com/posts/2"]
@pytest.mark.asyncio
async def test_multiple_rules_all_applied(self):
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
def rules(self):
return [
CrawlRule(LinkExtractor(allow=r"/posts/")),
CrawlRule(LinkExtractor(allow=r"/page/")),
]
spider = S()
out = await _collect(spider.parse(_make_response()))
urls = [r.url for r in out]
assert "https://example.com/posts/1" in urls
assert "https://example.com/posts/2" in urls
assert "https://example.com/page/2/" in urls
@pytest.mark.asyncio
async def test_rule_with_callback_bound_method(self):
spider = _TestSpider()
# rules() defaults to []; override at instance level
spider.rules = lambda: [ # type: ignore[method-assign]
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=spider.parse_post)
]
out = await _collect(spider.parse(_make_response()))
assert all(r.callback == spider.parse_post for r in out)
@pytest.mark.asyncio
async def test_rule_with_no_callback_leaves_request_callback_none(self):
# When CrawlRule.callback is None, response.follow() inherits the original
# request's callback. The original request was created with callback=None,
# so the resulting request's callback should also be None (engine then
# falls back to spider.parse).
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
spider = S()
out = await _collect(spider.parse(_make_response()))
assert all(r.callback is None for r in out)
@pytest.mark.asyncio
async def test_process_request_invoked(self):
spider = _TestSpider()
def add_priority(req: Request, response: Response) -> Request:
req.priority = 99
return req
spider.rules = lambda: [ # type: ignore[method-assign]
CrawlRule(LinkExtractor(allow=r"/posts/"), process_request=add_priority)
]
out = await _collect(spider.parse(_make_response()))
assert all(r.priority == 99 for r in out)
@pytest.mark.asyncio
async def test_process_request_can_replace_request(self):
spider = _TestSpider()
replacement = Request("https://replaced.example.com/")
def replace(req: Request, response: Response) -> Request:
return replacement
spider.rules = lambda: [ # type: ignore[method-assign]
CrawlRule(LinkExtractor(allow=r"/posts/"), process_request=replace)
]
out = await _collect(spider.parse(_make_response()))
assert all(r is replacement for r in out)
@pytest.mark.asyncio
async def test_user_can_compose_super_parse(self):
"""Override parse() to add custom yields plus call super().parse() for rules."""
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
async def parse(self, response):
yield {"custom": "item"}
async for req in super().parse(response):
yield req
spider = S()
out = await _collect(spider.parse(_make_response()))
assert out[0] == {"custom": "item"}
assert all(isinstance(x, Request) for x in out[1:])
assert len(out) == 3 # 1 dict + 2 requests
@pytest.mark.asyncio
async def test_referer_set_on_followed_requests(self):
# `response.follow()` sets the referer header; verify it survives the rule path.
class S(CrawlSpider):
name = "s"
start_urls = ["https://example.com/"]
def rules(self):
return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
spider = S()
out = await _collect(spider.parse(_make_response()))
for req in out:
assert req._session_kwargs["headers"]["referer"] == "https://example.com/"
class TestCrawlSpiderPickle:
"""Verify Request produced by CrawlSpider survives pickle round-trip with bound-method callbacks."""
@pytest.mark.asyncio
async def test_pickle_request_with_bound_method_callback(self):
spider = _TestSpider()
spider.rules = lambda: [ # type: ignore[method-assign]
CrawlRule(LinkExtractor(allow=r"/posts/"), callback=spider.parse_post)
]
out = await _collect(spider.parse(_make_response()))
req = out[0]
# __getstate__ should convert the bound method into a name-string
state = req.__getstate__()
assert state["callback"] is None
assert state["_callback_name"] == "parse_post"
# Round-trip via pickle (bound methods aren't directly picklable; the
# state machinery handles the conversion)
pickled = pickle.dumps(req)
restored = pickle.loads(pickled)
assert restored._callback_name == "parse_post"
# Then _restore_callback on a fresh spider instance brings the method back
fresh_spider = _TestSpider()
restored._restore_callback(fresh_spider)
assert restored.callback == fresh_spider.parse_post
class TestCrawlRule:
def test_default_callback_is_none(self):
rule = CrawlRule(LinkExtractor())
assert rule.callback is None
assert rule.priority is None
assert rule.process_request is None
def test_callback_accepts_callable(self):
spider = _TestSpider()
rule = CrawlRule(LinkExtractor(), callback=spider.parse_post)
assert rule.callback == spider.parse_post
+2 -2
View File
@@ -10,8 +10,8 @@ envlist = pre-commit,py{310,311,312,313}
usedevelop = True
changedir = tests
deps =
playwright==1.58.0
patchright==1.58.2
playwright==1.59.0
patchright==1.59.1
-r{toxinidir}/tests/requirements.txt
extras = ai,shell
commands =
+2
View File
@@ -35,6 +35,7 @@ nav = [
{"Requests & Responses" = "spiders/requests-responses.md"},
{"Sessions" = "spiders/sessions.md"},
{"Proxy management & Blocking" = "spiders/proxy-blocking.md"},
{"Generic crawlers" = "spiders/generic-templates.md"},
{"Advanced features" = "spiders/advanced.md"}
]},
{"Command Line Interface" = [
@@ -90,6 +91,7 @@ features = [
"search.share",
"search.suggest",
"search.highlight",
"content.code.copy",
]
[[project.theme.palette]]