Merge pull request #3 from D4Vinci/dev

Releasing version 0.2 (BIG CHANGES)
This commit is contained in:
Karim shoair
2024-11-11 19:29:41 +02:00
committed by GitHub
44 changed files with 2498 additions and 302 deletions
+30 -4
View File
@@ -7,15 +7,12 @@ concurrency:
jobs:
tests:
timeout-minutes: 60
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
include:
- python-version: "3.7"
os: ubuntu-latest
env:
TOXENV: py
- python-version: "3.8"
os: ubuntu-latest
env:
@@ -36,13 +33,42 @@ jobs:
os: ubuntu-latest
env:
TOXENV: py
- python-version: "3.13"
os: ubuntu-latest
env:
TOXENV: py
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
cache-dependency-path: |
setup.py
requirements*.txt
tox.ini
- name: Install Camoufox Dependencies
run: |
python3 -m pip install --upgrade pip
python3 -m pip install playwright camoufox
python3 -m playwright install chromium
python3 -m playwright install-deps chromium firefox
python3 -m camoufox fetch --browserforge
# Cache tox environments
- name: Cache tox environments
uses: actions/cache@v3
with:
path: .tox
# Include python version and os in cache key
key: tox-v1-${{ runner.os }}-py${{ matrix.python-version }}-${{ hashFiles('tox.ini', 'setup.py', 'requirements*.txt') }}
restore-keys: |
tox-v1-${{ runner.os }}-py${{ matrix.python-version }}-
tox-v1-${{ runner.os }}-
- name: Run tests
env: ${{ matrix.env }}
+2
View File
@@ -13,6 +13,8 @@ __pycache__/
.bootstrap
.appveyor.token
*.bak
*.db
*.db-*
# installation package
*.egg-info/
+7 -2
View File
@@ -15,7 +15,7 @@ configfile: pytest.ini
plugins: cov-5.0.0, anyio-4.6.0
collected 16 items
tests/test_all_functions.py ................ [100%]
tests/test_parser_functions.py ................ [100%]
=============================== 16 passed in 0.22s ================================
```
@@ -27,4 +27,9 @@ Also, consider setting `debug` to `True` while initializing the Adaptor object s
- Fork Scrapling [git repository](https://github.com/D4Vinci/Scrapling).
- Make your changes.
- Ensure tests work.
- Create a Pull Request against the [**dev**](https://github.com/D4Vinci/Scrapling/tree/dev) branch of Scrapling.
- Create a Pull Request against the [**dev**](https://github.com/D4Vinci/Scrapling/tree/dev) branch of Scrapling.
### Installing the latest changes from the dev branch
```commandline
pip3 install git+https://github.com/D4Vinci/Scrapling.git@dev
```
+2
View File
@@ -1,6 +1,8 @@
include LICENSE
include *.db
include *.js
include scrapling/*.db
include scrapling/*.db*
include scrapling/py.typed
recursive-exclude * __pycache__
+397 -73
View File
@@ -1,30 +1,77 @@
# 🕷️ Scrapling: Lightning-Fast, Adaptive Web Scraping for Python
# 🕷️ Scrapling: Undetectable, Lightning-Fast, and Adaptive Web Scraping for Python
[![Tests](https://github.com/D4Vinci/Scrapling/actions/workflows/tests.yml/badge.svg)](https://github.com/D4Vinci/Scrapling/actions/workflows/tests.yml) [![PyPI version](https://badge.fury.io/py/Scrapling.svg)](https://badge.fury.io/py/Scrapling) [![Supported Python versions](https://img.shields.io/pypi/pyversions/scrapling.svg)](https://pypi.org/project/scrapling/) [![PyPI Downloads](https://static.pepy.tech/badge/scrapling)](https://pepy.tech/project/scrapling)
Dealing with failing web scrapers due to website changes? Meet Scrapling.
Dealing with failing web scrapers due to anti-bot protections or website changes? Meet Scrapling.
Scrapling is a high-performance, intelligent web scraping library for Python that automatically adapts to website changes while significantly outperforming popular alternatives. Whether you're a beginner or an expert, Scrapling provides powerful features while maintaining simplicity.
Scrapling is a high-performance, intelligent web scraping library for Python that automatically adapts to website changes while significantly outperforming popular alternatives. For both beginners and experts, Scrapling provides powerful features while maintaining simplicity.
```python
from scrapling import Adaptor
# Scrape data that survives website changes
page = Adaptor(html, auto_match=True)
products = page.css('.product', auto_save=True)
# Later, even if selectors change:
products = page.css('.product', auto_match=True) # Still finds them!
>> from scrapling import Fetcher, StealthyFetcher, PlayWrightFetcher
# Fetch websites' source under the radar!
>> fetcher = StealthyFetcher().fetch('https://example.com', headless=True, disable_resources=True)
>> print(fetcher.status)
200
>> page = fetcher.adaptor
>> products = page.css('.product', auto_save=True) # Scrape data that survives website design changes!
>> # Later, if the website structure changes, pass `auto_match=True`
>> products = page.css('.product', auto_match=True) # and Scrapling still finds them!
```
## Table of content
* [Key Features](#key-features)
* [Fetch websites as you prefer](#fetch-websites-as-you-prefer)
* [Adaptive Scraping](#adaptive-scraping)
* [Performance](#performance)
* [Developing Experience](#developing-experience)
* [Getting Started](#getting-started)
* [Parsing Performance](#parsing-performance)
* [Text Extraction Speed Test (5000 nested elements).](#text-extraction-speed-test-5000-nested-elements)
* [Extraction By Text Speed Test](#extraction-by-text-speed-test)
* [Installation](#installation)
* [Fetching Websites Features](#fetching-websites-features)
* [Fetcher](#fetcher)
* [StealthyFetcher](#stealthyfetcher)
* [PlayWrightFetcher](#playwrightfetcher)
* [Advanced Parsing Features](#advanced-parsing-features)
* [Smart Navigation](#smart-navigation)
* [Content-based Selection & Finding Similar Elements](#content-based-selection--finding-similar-elements)
* [Handling Structural Changes](#handling-structural-changes)
* [Real World Scenario](#real-world-scenario)
* [Find elements by filters](#find-elements-by-filters)
* [Is That All?](#is-that-all)
* [More Advanced Usage](#more-advanced-usage)
* [⚡ Enlightening Questions and FAQs](#-enlightening-questions-and-faqs)
* [How does auto-matching work?](#how-does-auto-matching-work)
* [How does the auto-matching work if I didn't pass a URL while initializing the Adaptor object?](#how-does-the-auto-matching-work-if-i-didnt-pass-a-url-while-initializing-the-adaptor-object)
* [If all things about an element can change or get removed, what are the unique properties to be saved?](#if-all-things-about-an-element-can-change-or-get-removed-what-are-the-unique-properties-to-be-saved)
* [I have enabled the `auto_save`/`auto_match` parameter while selecting and it got completely ignored with a warning message](#i-have-enabled-the-auto_saveauto_match-parameter-while-selecting-and-it-got-completely-ignored-with-a-warning-message)
* [I have done everything as the docs but the auto-matching didn't return anything, what's wrong?](#i-have-done-everything-as-the-docs-but-the-auto-matching-didnt-return-anything-whats-wrong)
* [Can Scrapling replace code built on top of BeautifulSoup4?](#can-scrapling-replace-code-built-on-top-of-beautifulsoup4)
* [Can Scrapling replace code built on top of AutoScraper?](#can-scrapling-replace-code-built-on-top-of-autoscraper)
* [Is Scrapling thread-safe?](#is-scrapling-thread-safe)
* [Sponsors](#sponsors)
* [Contributing](#contributing)
* [Disclaimer for Scrapling Project](#disclaimer-for-scrapling-project)
* [License](#license)
* [Acknowledgments](#acknowledgments)
* [Thanks and References](#thanks-and-references)
* [Known Issues](#known-issues)
## Key Features
### Fetch websites as you prefer
- **HTTP requests**: Stealthy and fast HTTP requests with `Fetcher`
- **Stealthy fetcher**: Annoying anti-bot protection? No problem! Scrapling can bypass almost all of them with `StealthyFetcher` with default configuration!
- **Your preferred browser**: Use your real browser with CDP, [NSTbrowser](https://app.nstbrowser.io/r/1vO5e5)'s browserless, PlayWright with stealth mode, or even vanilla PlayWright - All is possible with `PlayWrightFetcher`!
### Adaptive Scraping
- 🔄 **Smart Element Tracking**: Locate previously identified elements after website structure changes, using an intelligent similarity system and integrated storage.
- 🎯 **Flexible Querying**: Use CSS selectors, XPath, text search, or regex - chain them however you want!
- 🎯 **Flexible Querying**: Use CSS selectors, XPath, Elements filters, text search, or regex - chain them however you want!
- 🔍 **Find Similar Elements**: Automatically locate elements similar to the element you want on the page (Ex: other products like the product you found on the page).
- 🧠 **Smart Content Scraping**: Extract data from multiple websites without specific selectors using its powerful features.
- 🧠 **Smart Content Scraping**: Extract data from multiple websites without specific selectors using Scrapling powerful features.
### Performance
- 🚀 **Lightning Fast**: Built from the ground up with performance in mind, outperforming most popular Python scraping libraries (outperforming BeautifulSoup by up to 237x in our tests).
- 🚀 **Lightning Fast**: Built from the ground up with performance in mind, outperforming most popular Python scraping libraries (outperforming BeautifulSoup in parsing by up to 620x in our tests).
- 🔋 **Memory Efficient**: Optimized data structures for minimal memory footprint.
-**Fast JSON serialization**: 10x faster JSON serialization than the standard json library with more options.
@@ -32,23 +79,18 @@ products = page.css('.product', auto_match=True) # Still finds them!
- 🛠️ **Powerful Navigation API**: Traverse the DOM tree easily in all directions and get the info you want (parent, ancestors, sibling, children, next/previous element, and more).
- 🧬 **Rich Text Processing**: All strings have built-in methods for regex matching, cleaning, and more. All elements' attributes are read-only dictionaries that are faster than standard dictionaries with added methods.
- 📝 **Automatic Selector Generation**: Create robust CSS/XPath selectors for any element.
- 🔌 **Scrapy-Compatible API**: Familiar methods and similar pseudo-elements for Scrapy users.
- 📘 **Type hints**: Complete type coverage for better IDE support and fewer bugs.
- 🔌 **API Similar to Scrapy/BeautifulSoup**: Familiar methods and similar pseudo-elements for Scrapy and BeautifulSoup users.
- 📘 **Type hints and test coverage**: Complete type coverage and almost full test coverage for better IDE support and fewer bugs, respectively.
## Getting Started
Let's walk through a basic example that demonstrates a small group of Scrapling's core features:
```python
import requests
from scrapling import Adaptor
from scrapling import Fetcher
# Fetch a web page
url = 'https://quotes.toscrape.com/'
response = requests.get(url)
fetcher = Fetcher(auto_match=False)
# Create an Adaptor instance
page = Adaptor(response.text, url=url)
# Fetch a web page and create an Adaptor instance
page = fetcher.get('https://quotes.toscrape.com/', stealthy_headers=True).adaptor
# Get all strings in the full page
page.get_all_text(ignore_tags=('script', 'style'))
@@ -56,10 +98,17 @@ page.get_all_text(ignore_tags=('script', 'style'))
quotes = page.css('.quote .text::text') # CSS selector
quotes = page.xpath('//span[@class="text"]/text()') # XPath
quotes = page.css('.quote').css('.text::text') # Chained selectors
quotes = [element.text for element in page.css('.quote').css('.text')] # Slower than bulk query above
quotes = [element.text for element in page.css('.quote .text')] # Slower than bulk query above
# Get the first quote element
quote = page.css('.quote').first # or [0] or .get()
quote = page.css_first('.quote') # / page.css('.quote').first / page.css('.quote')[0]
# Tired of selectors? Use find_all/find
quotes = page.find_all('div', {'class': 'quote'})
# Same as
quotes = page.find_all('div', class_='quote')
quotes = page.find_all(['div'], class_='quote')
quotes = page.find_all(class_='quote') # and so on...
# Working with elements
quote.html_content # Inner HTML
@@ -67,19 +116,9 @@ quote.prettify() # Prettified version of Inner HTML
quote.attrib # Element attributes
quote.path # DOM path to element (List)
```
To keep it simple, all methods can be chained on top of each other as long as you are chaining methods that return an element (It's called an `Adaptor` object) or a List of Adaptors (It's called `Adaptors` object)
To keep it simple, all methods can be chained on top of each other!
### Installation
Scrapling is a breeze to get started with - We only require at least Python 3.7 to work and the rest of the requirements are installed automatically with the package.
```bash
# Using pip
pip install scrapling
# Or the latest from GitHub
pip install git+https://github.com/D4Vinci/Scrapling.git@master
```
## Performance
## Parsing Performance
Scrapling isn't just powerful - it's also blazing fast. Scrapling implements many best practices, design patterns, and numerous optimizations to save fractions of seconds. All of that while focusing exclusively on parsing HTML documents.
Here are benchmarks comparing Scrapling to popular Python libraries in two tests.
@@ -106,11 +145,150 @@ As you see, Scrapling is on par with Scrapy and slightly faster than Lxml which
| Scrapling | 2.51 | 1.0x |
| AutoScraper | 11.41 | 4.546x |
Scrapling can find elements with more methods and it returns full element `Adaptor` objects not only the text like AutoScraper. So, to make this test fair, both libraries will extract an element with text, find similar elements, and then extract the text content for all of them. As you see, Scrapling is still 4.5 times faster at same task.
Scrapling can find elements with more methods and it returns full element `Adaptor` objects not only the text like AutoScraper. So, to make this test fair, both libraries will extract an element with text, find similar elements, and then extract the text content for all of them. As you see, Scrapling is still 4.5 times faster at the same task.
> All benchmarks' results are an average of 100 runs. See our [benchmarks.py](https://github.com/D4Vinci/Scrapling/blob/main/benchmarks.py) for methodology and to run your comparisons.
## Advanced Features
## Installation
Scrapling is a breeze to get started with - Starting from version 0.2, we require at least Python 3.8 to work.
```bash
pip3 install scrapling
```
- For using the `StealthyFetcher`, go to the command line and download the browser with
<details><summary>Windows OS</summary>
```bash
camoufox fetch --browserforge
```
</details>
<details><summary>MacOS</summary>
```bash
python3 -m camoufox fetch --browserforge
```
</details>
<details><summary>Linux</summary>
```bash
python -m camoufox fetch --browserforge
```
On a fresh installation of Linux, you may also need the following Firefox dependencies:
- Debian-based distros
```bash
sudo apt install -y libgtk-3-0 libx11-xcb1 libasound2
```
- Arch-based distros
```bash
sudo pacman -S gtk3 libx11 libxcb cairo libasound alsa-lib
```
</details>
<small> See the official <a href="https://camoufox.com/python/installation/#download-the-browser">Camoufox documentation</a> for more info on installation</small>
- If you are going to use the `PlayWrightFetcher` options, then install Playwright's Chromium browser with:
```commandline
playwright install chromium
```
- If you are going to use normal requests only with the `Fetcher` class then update the fingerprints files with:
```commandline
python -m browserforge update
```
## Fetching Websites Features
All fetcher-type classes are imported in the same way
```python
from scrapling import Fetcher, StealthyFetcher, PlayWrightFetcher
```
And all of them can take these initialization arguments: `auto_match`, `huge_tree`, `keep_comments`, `storage`, `storage_args`, and `debug` which are the same ones you give to the `Adaptor` class.
> [!NOTE]
> The `auto_match` argument is enabled by default which is the one you should care about the most as you will see later.
### Fetcher
This class is built on top of [httpx](https://www.python-httpx.org/) with additional configuration options, here you can do `GET`, `POST`, `PUT`, and `DELETE` requests.
For all methods, you have `stealth_headers` which makes `Fetcher` create and use real browser's headers then create a referer header as if this request came from Google's search of this URL's domain. It's enabled by default.
```python
>> page = Fetcher().get('https://httpbin.org/get', stealth_headers=True, follow_redirects=True)
>> page = Fetcher().post('https://httpbin.org/post', data={'key': 'value'})
>> page = Fetcher().put('https://httpbin.org/put', data={'key': 'value'})
>> page = Fetcher().delete('https://httpbin.org/delete')
```
### StealthyFetcher
This class is built on top of [Camoufox](https://github.com/daijro/camoufox) which by default bypasses most of the anti-bot protections. Scrapling adds extra layers of flavors and configurations to increase performance and undetectability even further.
```python
>> page = StealthyFetcher().fetch('https://www.browserscan.net/bot-detection') # Running headless by default
>> page.status == 200
True
```
<details><summary><strong>For the sake of simplicity, expand this for the complete list of arguments</strong></summary>
| Argument | Description | Optional |
|:-------------------:|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:--------:|
| url | Target url | ❌ |
| headless | Pass `True` to run the browser in headless/hidden (**default**), `virtual` to run it in virtual screen mode, or `False` for headful/visible mode. The `virtual` mode requires having `xvfb` installed. | ✔️ |
| block_images | Prevent the loading of images through Firefox preferences. _This can help save your proxy usage but be careful with this option as it makes some websites never finish loading._ | ✔️ |
| disable_resources | Drop requests of unnecessary resources for a speed boost. It depends but it made requests ~25% faster in my tests for some websites.<br/>Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`. _This can help save your proxy usage but be careful with this option as it makes some websites never finish loading._ | ✔️ |
| google_search | Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name. | ✔️ |
| extra_headers | A dictionary of extra headers to add to the request. _The referer set by the `google_search` argument takes priority over the referer set here if used together._ | ✔️ |
| block_webrtc | Blocks WebRTC entirely. | ✔️ |
| page_action | Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again. | ✔️ |
| addons | List of Firefox addons to use. **Must be paths to extracted addons.** | ✔️ |
| humanize | Humanize the cursor movement. Takes either True or the MAX duration in seconds of the cursor movement. The cursor typically takes up to 1.5 seconds to move across the window. | ✔️ |
| allow_webgl | Whether to allow WebGL. To prevent leaks, only use this for special cases. | ✔️ |
| network_idle | Wait for the page until there are no network connections for at least 500 ms. | ✔️ |
| timeout | The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000. | ✔️ |
| wait_selector | Wait for a specific css selector to be in a specific state. | ✔️ |
| wait_selector_state | The state to wait for the selector given with `wait_selector`. _Default state is `attached`._ | ✔️ |
</details>
This list isn't final so expect a lot more additions and flexibility to be added in the next versions!
### PlayWrightFetcher
This class is built on top of [Playwright](https://playwright.dev/python/) which currently provides 4 main run options but they can be mixed as you want.
```python
>> page = PlayWrightFetcher().fetch('https://www.google.com/search?q=%22Scrapling%22', disable_resources=True) # Vanilla Playwright option
>> page.adaptor.css_first("#search a::attr(href)")
'https://github.com/D4Vinci/Scrapling'
```
Using this Fetcher class, you can make requests with:
1) Vanilla Playwright without any modifications other than the ones you chose.
2) Stealthy Playwright with the stealth mode I wrote for it. It's still a WIP but it bypasses many online tests like [Sannysoft's](https://bot.sannysoft.com/).</br> Some of the things this fetcher's stealth mode does include:
* Patching the CDP runtime fingerprint.
* Mimics some of the real browsers' properties by injecting several JS files and using custom options.
* Using custom flags on launch to hide Playwright even more and make it faster.
* Generates real browser's headers of the same type and same user OS then append it to the request's headers.
3) Real browsers by passing the CDP URL of your browser to be controlled by the Fetcher and most of the options can be enabled on it.
4) [NSTBrowser](https://app.nstbrowser.io/r/1vO5e5)'s [docker browserless](https://hub.docker.com/r/nstbrowser/browserless) option by passing the CDP URL and enabling `nstbrowser_mode` option.
Add that to a lot of controlling/hiding options as you will see in the arguments list below.
<details><summary><strong>Expand this for the complete list of arguments</strong></summary>
| Argument | Description | Optional |
|:-------------------:|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:--------:|
| url | Target url | ❌ |
| headless | Pass `True` to run the browser in headless/hidden (**default**), or `False` for headful/visible mode. | ✔️ |
| disable_resources | Drop requests of unnecessary resources for a speed boost. It depends but it made requests ~25% faster in my tests for some websites.<br/>Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`. _This can help save your proxy usage but be careful with this option as it makes some websites never finish loading._ | ✔️ |
| useragent | Pass a useragent string to be used. **Otherwise the fetcher will generate a real Useragent of the same browser and use it.** | ✔️ |
| network_idle | Wait for the page until there are no network connections for at least 500 ms. | ✔️ |
| timeout | The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000. | ✔️ |
| page_action | Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again. | ✔️ |
| wait_selector | Wait for a specific css selector to be in a specific state. | ✔️ |
| wait_selector_state | The state to wait for the selector given with `wait_selector`. _Default state is `attached`._ | ✔️ |
| google_search | Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name. | ✔️ |
| extra_headers | A dictionary of extra headers to add to the request. The referer set by the `google_search` argument takes priority over the referer set here if used together. | ✔️ |
| hide_canvas | Add random noise to canvas operations to prevent fingerprinting. | ✔️ |
| disable_webgl | Disables WebGL and WebGL 2.0 support entirely. | ✔️ |
| stealth | Enables stealth mode, always check the documentation to see what stealth mode does currently. | ✔️ |
| cdp_url | Instead of launching a new browser instance, connect to this CDP URL to control real browsers/NSTBrowser through CDP. | ✔️ |
| nstbrowser_mode | Enables NSTBrowser mode, **it have to be used with `cdp_url` argument or it will get completely ignored.** | ✔️ |
| nstbrowser_config | The config you want to send with requests to the NSTBrowser. _If left empty, Scrapling defaults to an optimized NSTBrowser's docker browserless config._ | ✔️ |
</details>
This list isn't final so expect a lot more additions and flexibility to be added in the next versions!
## Advanced Parsing Features
### Smart Navigation
```python
>>> quote.tag
@@ -130,24 +308,23 @@ Scrapling can find elements with more methods and it returns full element `Adapt
>>> quote.siblings
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
...]
>>> quote.next # gets the next element, the same logic applies to `quote.previous`
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>
>>> quote.children.css(".author::text")
['Albert Einstein']
>>> quote.children.css_first(".author::text")
'Albert Einstein'
>>> quote.has_class('quote')
True
# Generate new selectors for any element
>>> quote.css_selector
>>> quote.generate_css_selector
'body > div > div:nth-of-type(2) > div > div'
# Test these selectors on your favorite browser or reuse them again in the library in other methods!
>>> quote.xpath_selector
# Test these selectors on your favorite browser or reuse them again in the library's methods!
>>> quote.generate_xpath_selector
'//body/div/div[2]/div/div'
```
If your case needs more than the element's parent, you can iterate over the whole ancestors' tree of any element like below
@@ -164,11 +341,9 @@ You can search for a specific ancestor of an element that satisfies a function,
### Content-based Selection & Finding Similar Elements
You can select elements by their text content in multiple ways, here's a full example on another website:
```python
>>> response = requests.get('https://books.toscrape.com/index.html')
>>> page = Fetcher().get('https://books.toscrape.com/index.html').adaptor
>>> page = Adaptor(response.text, url=response.url)
>>> page.find_by_text('Tipping the Velvet') # Find the first element that its text fully matches this text
>>> page.find_by_text('Tipping the Velvet') # Find the first element whose text fully matches this text
<data='<a href="catalogue/tipping-the-velvet_99...' parent='<h3><a href="catalogue/tipping-the-velve...'>
>>> page.find_by_text('Tipping the Velvet', first_match=False) # Get all matches if there are more
@@ -208,8 +383,8 @@ To increase the complexity a little bit, let's say we want to get all books' dat
```python
>>> for product in page.find_by_text('Tipping the Velvet').parent.parent.find_similar():
print({
"name": product.css('h3 a::text')[0],
"price": product.css('.price_color')[0].re_first(r'[\d\.]+'),
"name": product.css_first('h3 a::text'),
"price": product.css_first('.price_color').re_first(r'[\d\.]+'),
"stock": product.css('.availability::text')[-1].clean()
})
{'name': 'A Light in the ...', 'price': '51.77', 'stock': 'In stock'}
@@ -220,8 +395,6 @@ To increase the complexity a little bit, let's say we want to get all books' dat
The [documentation](https://github.com/D4Vinci/Scrapling/tree/main/docs/Examples) will provide more advanced examples.
### Handling Structural Changes
> Because [the internet archive](https://web.archive.org/) is down at the time of writing this, I can't use real websites as examples even though I tested that before (I mean browsing an old version of a website and then counting the current version of the website as structural changes)
Let's say you are scraping a page with a structure like this:
```html
<div class="container">
@@ -237,7 +410,7 @@ Let's say you are scraping a page with a structure like this:
</section>
</div>
```
and you want to scrape the first product, the one with the `p1` ID. You will probably write a selector like this
And you want to scrape the first product, the one with the `p1` ID. You will probably write a selector like this
```python
page.css('#p1')
```
@@ -262,34 +435,147 @@ When website owners implement structural changes like
</div>
</div>
```
The selector will no longer function and your code needs maintenance. That's where Scrapling auto-matching feature comes into play.
The selector will no longer function and your code needs maintenance. That's where Scrapling's auto-matching feature comes into play.
```python
from scrapling import Adaptor
# Before the change
page = Adaptor(page_source, url='example.com', auto_match=True)
page = Adaptor(page_source, url='example.com')
element = page.css('#p1' auto_save=True)
if not element: # One day website changes?
element = page.css('#p1', auto_match=True) # Still finds it!
element = page.css('#p1', auto_match=True) # Scrapling still finds it!
# the rest of the code...
```
> How does the auto-matching work? Check the [FAQs](#FAQs) section for that and other possible issues while auto-matching.
> How does the auto-matching work? Check the [FAQs](#-enlightening-questions-and-faqs) section for that and other possible issues while auto-matching.
#### Real-World Scenario
Let's use a real website as an example and use one of the fetchers to fetch its source. To do this we need to find a website that will change its design/structure soon, take a copy of its source then wait for the website to make the change. Of course, that's nearly impossible to know unless I know the website's owner but that will make it a staged test haha.
To solve this issue, I will use [The Web Archive](https://archive.org/)'s [Wayback Machine](https://web.archive.org/). Here is a copy of [StackOverFlow's website in 2010](https://web.archive.org/web/20100102003420/http://stackoverflow.com/), pretty old huh?</br>Let's test if the automatch feature can extract the same button in the old design from 2010 and the current design using the same selector :)
If I want to extract the Questions button from the old design I can use a selector like this `#hmenus > div:nth-child(1) > ul > li:nth-child(1) > a` This selector is too specific because it was generated by Google Chrome.
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/"
>>
>> page = Fetcher(automatch_domain='stackoverflow.com').get(old_url, timeout=30).adaptor
>> element1 = page.css_first(selector, auto_save=True)
>>
>> # Same selector but used in the updated website
>> page = Fetcher(automatch_domain="stackoverflow.com").get(new_url).adaptor
>> element2 = page.css_first(selector, auto_match=True)
>>
>> if element1.text == element2.text:
... print('Scrapling found the same element in the old design and the new design!')
'Scrapling found the same element in the old design and the new design!'
```
Note that I used a new argument called `automatch_domain`, this is because for Scrapling these are two different URLs, not the website so it isolates their data. To tell Scrapling they are the same website, we then pass the domain we want to use for saving auto-match data for them both so Scrapling doesn't isolate them.
In a real-world scenario, the code will be the same except it will use the same URL for both requests so you won't need to use the `automatch_domain` argument. This is the closest example I can give to real-world cases so I hope it didn't confuse you :)
**Notes:**
1. Passing the `auto_save` argument without setting `auto_match` to `True` while initializing the Adaptor object will only result in ignoring the `auto_save` argument value and the following warning message
1. For the two examples above I used one time the `Adaptor` class and the second time the `Fetcher` class just to show you that you can create the `Adaptor` object by yourself if you have the source or fetch the source using any `Fetcher` class then it will create the `Adaptor` object for you on the `.adaptor` property.
2. Passing the `auto_save` argument with the `auto_match` argument set to `False` while initializing the Adaptor/Fetcher object will only result in ignoring the `auto_save` argument value and the following warning message
```text
Argument `auto_save` will be ignored because `auto_match` wasn't enabled on initialization. Check docs for more info.
```
This behavior is purely for performance reasons so the database gets created/connected only when you are planning to use the auto-matching features. Same case with the `auto_match` argument.
2. The `auto_match` parameter works only for `Adaptor` instances not `Adaptors` so if you do something like this you will get an error
3. The `auto_match` parameter works only for `Adaptor` instances not `Adaptors` so if you do something like this you will get an error
```python
page.css('body').css('#p1', auto_match=True)
```
because you can't auto-match a whole list, you have to be specific and do something like
```python
page.css('body')[0].css('#p1', auto_match=True)
page.css_first('body').css('#p1', auto_match=True)
```
### Find elements by filters
Inspired by BeautifulSoup's `find_all` function you can find elements by using `find_all`/`find` methods. Both methods can take multiple types of filters and return all elements in the pages that all these filters apply to.
* To be more specific:
* Any string passed is considered a tag name
* Any iterable passed like List/Tuple/Set is considered an iterable of tag names.
* Any dictionary is considered a mapping of HTML element(s) attribute names and attribute values.
* Any regex patterns passed are used as filters
* Any functions passed are used as filters
* Any keyword argument passed is considered as an HTML element attribute with its value.
So the way it works is after collecting all passed arguments and keywords, each filter passes its results to the following filter in a waterfall-like filtering system.
<br/>It filters all elements in the current page/element in the following order:
1. All elements with the passed tag name(s).
2. All elements that match all passed attribute(s).
3. All elements that match all passed regex patterns.
4. All elements that fulfill all passed function(s).
Note: The filtering process always starts from the first filter it finds in the filtering order above so if no tag name(s) are passed but attributes are passed, the process starts from that layer and so on. **But the order in which you pass the arguments doesn't matter.**
Examples to clear any confusion :)
```python
>> from scrapling import Fetcher
>> page = Fetcher().get('https://quotes.toscrape.com/').adaptor
# Find all elements with tag name `div`.
>> page.find_all('div')
[<data='<div class="container"> <div class="row...' parent='<body> <div class="container"> <div clas...'>,
<data='<div class="row header-box"> <div class=...' parent='<div class="container"> <div class="row...'>,
...]
# Find all div elements with a class that equals `quote`.
>> page.find_all('div', class_='quote')
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
...]
# Same as above.
>> page.find_all('div', {'class': 'quote'})
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
...]
# Find all elements with a class that equals `quote`.
>> page.find_all({'class': 'quote'})
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
...]
# Find all div elements with a class that equals `quote`, and contains the element `.text` which contains the word 'world' in its content.
>> page.find_all('div', {'class': 'quote'}, lambda e: "world" in e.css_first('.text::text'))
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>]
# Find all elements that don't have children.
>> page.find_all(lambda element: len(element.children) > 0)
[<data='<html lang="en"><head><meta charset="UTF...'>,
<data='<head><meta charset="UTF-8"><title>Quote...' parent='<html lang="en"><head><meta charset="UTF...'>,
<data='<body> <div class="container"> <div clas...' parent='<html lang="en"><head><meta charset="UTF...'>,
...]
# Find all elements that contain the word 'world' in its content.
>> page.find_all(lambda element: "world" in element.text)
[<data='<span class="text" itemprop="text">“The...' parent='<div class="quote" itemscope itemtype="h...'>,
<data='<a class="tag" href="/tag/world/page/1/"...' parent='<div class="tags"> Tags: <meta class="ke...'>]
# Find all span elements that match the given regex
>> page.find_all('span', re.compile(r'world'))
[<data='<span class="text" itemprop="text">“The...' parent='<div class="quote" itemscope itemtype="h...'>]
# Find all div and span elements with class 'quote' (No span elements like that so only div returned)
>> page.find_all(['div', 'span'], {'class': 'quote'})
[<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
<data='<div class="quote" itemscope itemtype="h...' parent='<div class="col-md-8"> <div class="quote...'>,
...]
# Mix things up
>> page.find_all({'itemtype':"http://schema.org/CreativeWork"}, 'div').css('.author::text')
['Albert Einstein',
'J.K. Rowling',
...]
```
### Is That All?
Here's what else you can do with Scrapling:
@@ -300,12 +586,12 @@ Here's what else you can do with Scrapling:
```
- Saving and retrieving elements manually to auto-match them outside the `css` and the `xpath` methods but you have to set the identifier by yourself.
- To save element to the database:
- To save an element to the database:
```python
>>> element = page.find_by_text('Tipping the Velvet', first_match=True)
>>> page.save(element, 'my_special_element')
```
- Now later when you want to retrieve it and relocate it in the page with auto-matching, it would be like this
- Now later when you want to retrieve it and relocate it inside the page with auto-matching, it would be like this
```python
>>> element_dict = page.retrieve('my_special_element')
>>> page.relocate(element_dict, adaptor_type=True)
@@ -319,13 +605,38 @@ Here's what else you can do with Scrapling:
[<Element a at 0x105a2a7b0>]
```
- Filtering results based on a function
```python
# Find all products over $50
expensive_products = page.css('.product_pod').filter(
lambda p: float(p.css('.price_color').re_first(r'[\d\.]+')) > 50
)
```
- Searching results for the first one that matches a function
```python
# Find all the products with price '53.23'
page.css('.product_pod').search(
lambda p: float(p.css('.price_color').re_first(r'[\d\.]+')) == 54.23
)
```
- Doing operations on element content is the same as scrapy
```python
quote.re(r'somethings') # Get all strings (TextHandlers) that match the regex pattern
quote.re_first(r'something') # Get the first string (TextHandler) only
quote.re(r'regex_pattern') # Get all strings (TextHandlers) that match the regex pattern
quote.re_first(r'regex_pattern') # Get the first string (TextHandler) only
quote.json() # If the content text is jsonable, then convert it to json using `orjson` which is 10x faster than the standard json library and provides more options
```
Hence all of these methods are actually methods from the `TextHandler` within that contains the text content so the same can be done directly if you call the `.text` property or equivalent selector function.
except that you can do more with them like
```python
quote.re(
r'regex_pattern',
replace_entities=True, # Character entity references are replaced by their corresponding character
clean_match=True, # This will ignore all whitespaces and consecutive spaces while matching
case_sensitive= False, # Set the regex to ignore letters case while compiling it
)
```
Hence all of these methods are methods from the `TextHandler` within that contains the text content so the same can be done directly if you call the `.text` property or equivalent selector function.
- Doing operations on the text content itself includes
@@ -339,11 +650,11 @@ Here's what else you can do with Scrapling:
```
- Sort all characters in the string as if it were a list and return the new string
```python
quote.sort()
quote.sort(reverse=False)
```
> To be clear, `TextHandler` is a sub-class of Python's `str` so all normal operations/methods that work with Python strings will work with it.
- Any element's attributes are not exactly a dictionary but a sub-class of [mapping](https://docs.python.org/3/glossary.html#term-mapping) called `AttributesHandler` that's read-only so it's faster and string values returned are actually `TextHandler` objects so all operations above can be done on them, standard dictionary operations that doesn't modify the data, and more :)
- Any element's attributes are not exactly a dictionary but a sub-class of [mapping](https://docs.python.org/3/glossary.html#term-mapping) called `AttributesHandler` that's read-only so it's faster and string values returned are actually `TextHandler` objects so all operations above can be done on them, standard dictionary operations that don't modify the data, and more :)
- Unlike standard dictionaries, here you can search by values too and can do partial searches. It might be handy in some cases (returns a generator of matches)
```python
>>> for item in element.attrib.search_values('catalogue', partial=True):
@@ -370,8 +681,9 @@ There are a lot of deep details skipped here to make this as short as possible s
Note that implementing your storage system can be complex as there are some strict rules such as inheriting from the same abstract class, following the singleton design pattern used in other classes, and more. So make sure to read the docs first.
To give detailed documentation of the library, it will need a website. I'm trying to rush creating the website, researching new ideas, and adding more features/tests/benchmarks but time is tight with too many spinning plates between work, personal life, and working on Scrapling. But you can help by using the [sponsor button](https://github.com/sponsors/D4Vinci) above :)
## FAQs
## ⚡ Enlightening Questions and FAQs
This section addresses common questions about Scrapling, please read this section before opening an issue.
### How does auto-matching work?
@@ -384,7 +696,7 @@ This section addresses common questions about Scrapling, please read this sectio
Together both are used to retrieve the element's unique properties from the database later.
4. Now later when you enable the `auto_match` parameter for both the Adaptor instance and the method call. The element properties are retrieved and Scrapling loops over all elements in the page and compares each one's unique properties to the unique properties we already have for this element and a score is calculated for each one.
5. The comparison between elements is not exact but more about finding how similar these values are, so everything is taken into consideration even the values' order like the order in which the element class names were written before and the order in which the same element class names are written now.
6. The score for each element is stored in the table and in the end, the element(s) with the highest combined similarity scores are returned.
6. The score for each element is stored in the table, and in the end, the element(s) with the highest combined similarity scores are returned.
### How does the auto-matching work if I didn't pass a URL while initializing the Adaptor object?
Not a big problem as it depends on your usage. The word `default` will be used in place of the URL field while saving the element's unique properties. So this will only be an issue if you used the same identifier later for a different website that you didn't pass the URL parameter while initializing it as well. The save process will overwrite the previous data and auto-matching uses the latest saved properties only.
@@ -413,7 +725,7 @@ Pretty much yeah, almost all features you get from BeautifulSoup can be found or
Of course, you can find elements by text/regex, find similar elements in a more reliable way than AutoScraper, and finally save/retrieve elements manually to use later as the model feature in AutoScraper. I have pulled all top articles about AutoScraper from Google and tested Scrapling against examples in them. In all examples, Scrapling got the same results as AutoScraper in much less time.
### Is Scrapling thread-safe?
Yes, Scrapling instances are thread-safe. Each Adaptor instance maintains its own state.
Yes, Scrapling instances are thread-safe. Each Adaptor instance maintains its state.
## Sponsors
[![Capsolver Banner](https://raw.githubusercontent.com/D4Vinci/Scrapling/main/images/CapSolver.png)](https://www.capsolver.com/?utm_source=github&utm_medium=repo&utm_campaign=scraping&utm_term=Scrapling)
@@ -423,6 +735,10 @@ Everybody is invited and welcome to contribute to Scrapling. There is a lot to d
Please read the [contributing file](https://github.com/D4Vinci/Scrapling/blob/main/CONTRIBUTING.md) before doing anything.
## Disclaimer for Scrapling Project
> [!CAUTION]
> This library is provided for educational and research purposes only. By using this library, you agree to comply with local and international laws regarding data scraping and privacy. The authors and contributors are not responsible for any misuse of this software. This library should not be used to violate the rights of others, for unethical purposes, or to use data in an unauthorized or illegal manner. Do not use it on any website unless you have permission from the website owner or within their allowed rules like the `robots.txt` file, for example.
## License
This work is licensed under BSD-3
@@ -430,8 +746,16 @@ This work is licensed under BSD-3
This project includes code adapted from:
- Parsel (BSD License) - Used for [translator](https://github.com/D4Vinci/Scrapling/blob/main/scrapling/translator.py) submodule
## Thanks and References
- [Daijro](https://github.com/daijro)'s brilliant work on both [BrowserForge](https://github.com/daijro/browserforge) and [Camoufox](https://github.com/daijro/camoufox)
- [Vinyzu](https://github.com/Vinyzu)'s work on Playwright's mock on [Botright](https://github.com/Vinyzu/Botright)
- [brotector](https://github.com/kaliiiiiiiiii/brotector)
- [fakebrowser](https://github.com/kkoooqq/fakebrowser)
- [rebrowser-patches](https://github.com/rebrowser/rebrowser-patches)
## Known Issues
- In the auto-matching save process, the unique properties of the first element from the selection results are the only ones that get saved. So if the selector you are using selects different elements on the page that are in different locations, auto-matching will probably return to you the first element only when you relocate it later. This doesn't include combined CSS selectors (Using commas to combine more than one selector for example) as these selectors get separated and each selector gets executed alone.
- Currently, Scrapling is not compatible with async/await.
<div align="center"><small>Made with ❤️ by Karim Shoair</small></div><br>
---
<div align="center"><small>Designed & crafted with ❤️ by Karim Shoair.</small></div><br>
+12 -11
View File
@@ -1,13 +1,14 @@
## TODOs
- Add more tests and increase the code coverage.
- Structure the tests folder in a better way.
- Add more documentation.
- Add the browsing ability.
- Create detailed documentation for 'readthedocs' website, preferably add Github action for deploying it.
- Create a Scrapy plugin/decorator to make it replace parsel in the response argument when needed.
- Need to add more functionality to `AttributesHandler` and more navigation functions to `Adaptor` object (ex: functions similar to map, filter, and reduce functions but here pass it to the element and the function is executed on children, siblings, next elements, etc...)
- Add `.filter` method to `Adaptors` object and other similar methods.
- Add functionality to automatically detect pagination URLs
- Add the ability to auto-detect schemas in pages and manipulate them
- Add ability to generate a regex from a group of elements (Like for all href attributes)
- [x] Add more tests and increase the code coverage.
- [x] Structure the tests folder in a better way.
- [ ] Add more documentation.
- [x] Add the browsing ability.
- [ ] Create detailed documentation for 'readthedocs' website, preferably add Github action for deploying it.
- [ ] Create a Scrapy plugin/decorator to make it replace parsel in the response argument when needed.
- [ ] Need to add more functionality to `AttributesHandler` and more navigation functions to `Adaptor` object (ex: functions similar to map, filter, and reduce functions but here pass it to the element and the function is executed on children, siblings, next elements, etc...)
- [x] Add `.filter` method to `Adaptors` object and other similar methods.
- [ ] Add functionality to automatically detect pagination URLs
- [ ] Add the ability to auto-detect schemas in pages and manipulate them.
- [ ] Add `analyzer` ability that tries to learn about the page through meta elements and return what it learned
- [ ] Add ability to generate a regex from a group of elements (Like for all href attributes)
-
+1 -1
View File
@@ -1,2 +1,2 @@
[pytest]
addopts = -p no:warnings --doctest-modules --ignore=setup.py
addopts = -p no:warnings --doctest-modules --ignore=setup.py --verbose
+4 -3
View File
@@ -1,10 +1,11 @@
# Declare top-level shortcuts
from scrapling.fetchers import Fetcher, StealthyFetcher, PlayWrightFetcher, CustomFetcher
from scrapling.parser import Adaptor, Adaptors
from scrapling.custom_types import TextHandler, AttributesHandler
from scrapling.core.custom_types import TextHandler, AttributesHandler
__author__ = "Karim Shoair (karim.shoair@pm.me)"
__version__ = "0.1.2"
__version__ = "0.2"
__copyright__ = "Copyright (c) 2024 Karim Shoair"
__all__ = ['Adaptor', 'Adaptors', 'TextHandler', 'AttributesHandler']
__all__ = ['Adaptor', 'Fetcher', 'StealthyFetcher', 'PlayWrightFetcher']
View File
+25
View File
@@ -0,0 +1,25 @@
"""
Type definitions for type checking purposes.
"""
from typing import (
Dict, Optional, Union, Callable, Any, List, Tuple, Pattern, Generator, Iterable, Type, TYPE_CHECKING, Literal
)
try:
from typing import Protocol
except ImportError:
# Added in Python 3.8
Protocol = object
try:
from typing import SupportsIndex
except ImportError:
# 'SupportsIndex' got added in Python 3.8
SupportsIndex = None
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
else:
Self = object
@@ -1,9 +1,9 @@
import re
from types import MappingProxyType
from collections.abc import Mapping
from typing import Dict, List, Union, Pattern
from scrapling.utils import _is_iterable, flatten
from scrapling.core.utils import _is_iterable, flatten
from scrapling.core._types import Dict, List, Union, Pattern, SupportsIndex
from orjson import loads, dumps
from w3lib.html import replace_entities as _replace_entities
@@ -69,7 +69,7 @@ class TextHandler(str):
return [TextHandler(_replace_entities(s)) for s in results]
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False,):
clean_match: bool = False, case_sensitive: bool = False) -> Union[str, None]:
"""Apply the given regex to text and return the first match if found, otherwise return the default value.
:param regex: Can be either a compiled regular expression or a string.
@@ -83,6 +83,51 @@ class TextHandler(str):
return result[0] if result else default
class TextHandlers(List[TextHandler]):
"""
The :class:`TextHandlers` class is a subclass of the builtin ``List`` class, which provides a few additional methods.
"""
__slots__ = ()
def __getitem__(self, pos: Union[SupportsIndex, slice]) -> Union[TextHandler, "TextHandlers[TextHandler]"]:
lst = super().__getitem__(pos)
if isinstance(pos, slice):
return self.__class__(lst)
else:
return lst
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True, clean_match: bool = False,
case_sensitive: bool = False) -> 'List[str]':
"""Call the ``.re()`` method for each element in this list and return
their results flattened as TextHandlers.
:param regex: Can be either a compiled regular expression or a string.
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
results = [
n.re(regex, replace_entities, clean_match, case_sensitive) for n in self
]
return flatten(results)
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False) -> Union[str, None]:
"""Call the ``.re_first()`` method for each element in this list and return
the first result or the default value otherwise.
:param regex: Can be either a compiled regular expression or a string.
:param default: The default value to be returned if there is no match
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
for n in self:
for result in n.re(regex, replace_entities, clean_match, case_sensitive):
return result
return default
class AttributesHandler(Mapping):
"""A read-only mapping to use instead of the standard dictionary for the speed boost but
at the same time I use it to add more functionalities.
@@ -4,7 +4,7 @@ class SelectorsGeneration:
Trying to generate selectors like Firefox or maybe cleaner ones!? Ehm
Inspiration: https://searchfox.org/mozilla-central/source/devtools/shared/inspector/css-logic.js#591"""
def __general_selection(self, selection: str = 'css') -> str:
def __general_selection(self, selection: str = 'css', full_path=False) -> str:
"""Generate a selector for the current element.
:return: A string of the generated selector.
"""
@@ -20,10 +20,11 @@ class SelectorsGeneration:
else f"[@id='{target.attrib['id']}']"
)
selectorPath.append(part)
return (
" > ".join(reversed(selectorPath)) if css
else '//*' + "/".join(reversed(selectorPath))
)
if not full_path:
return (
" > ".join(reversed(selectorPath)) if css
else '//*' + "/".join(reversed(selectorPath))
)
else:
part = f'{target.tag}'
# We won't use classes anymore because I some websites share exact classes between elements
@@ -60,15 +61,29 @@ class SelectorsGeneration:
)
@property
def css_selector(self) -> str:
def generate_css_selector(self) -> str:
"""Generate a CSS selector for the current element
:return: A string of the generated selector.
"""
return self.__general_selection()
@property
def xpath_selector(self) -> str:
def generate_full_css_selector(self) -> str:
"""Generate a complete CSS selector for the current element
:return: A string of the generated selector.
"""
return self.__general_selection(full_path=True)
@property
def generate_xpath_selector(self) -> str:
"""Generate a XPath selector for the current element
:return: A string of the generated selector.
"""
return self.__general_selection('xpath')
@property
def generate_full_xpath_selector(self) -> str:
"""Generate a complete XPath selector for the current element
:return: A string of the generated selector.
"""
return self.__general_selection('xpath', full_path=True)
@@ -4,9 +4,9 @@ import logging
import threading
from hashlib import sha256
from abc import ABC, abstractmethod
from typing import Dict, Optional, Union
from scrapling.utils import _StorageTools, cache
from scrapling.core._types import Dict, Optional, Union
from scrapling.core.utils import _StorageTools, cache
from lxml import html
from tldextract import extract as tld
@@ -9,24 +9,14 @@ which will be important in future releases but most importantly...
import re
from w3lib.html import HTML5_WHITESPACE
from typing import TYPE_CHECKING, Any, Optional
try:
from typing import Protocol
except ImportError:
# Added in Python 3.8
Protocol = object
from scrapling.utils import cache
from scrapling.core.utils import cache
from scrapling.core._types import Any, Optional, Protocol, Self
from cssselect.xpath import ExpressionError
from cssselect.xpath import XPathExpr as OriginalXPathExpr
from cssselect import HTMLTranslator as OriginalHTMLTranslator
from cssselect.parser import Element, FunctionalPseudoElement, PseudoElement
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
regex = f"[{HTML5_WHITESPACE}]+"
replace_html5_whitespaces = re.compile(regex).sub
+2 -61
View File
@@ -1,14 +1,13 @@
import re
import os
import logging
from itertools import chain
from logging import handlers
# Using cache on top of a class is brilliant way to achieve Singleton design pattern without much code
from functools import lru_cache as cache # functools.cache is available on Python 3.9+ only so let's keep lru_cache
from typing import Dict, Iterable, Any
from scrapling.core._types import Dict, Iterable, Any
from lxml import html
html_forbidden = {html.HtmlComment, }
logging.basicConfig(
level=logging.ERROR,
@@ -45,64 +44,6 @@ def _is_iterable(s: Any):
return isinstance(s, (list, tuple,))
@cache(None, typed=True)
class _Logger(object):
# I will leave this class here for now in case I decide I want to come back to use it :)
__slots__ = ('console_logger', 'logger_file_path',)
levels = {
'debug': logging.DEBUG,
'info': logging.INFO,
'warning': logging.WARNING,
'error': logging.ERROR,
'critical': logging.CRITICAL
}
def __init__(self, filename: str = 'debug.log', level: str = 'debug', when: str = 'midnight', backcount: int = 1):
os.makedirs(os.path.join(os.path.dirname(__file__), 'logs'), exist_ok=True)
format_str = logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s", "%Y-%m-%d %H:%M:%S")
# on-screen output
lvl = self.levels[level.lower()]
self.console_logger = logging.getLogger('Scrapling')
self.console_logger.setLevel(lvl)
console_handler = logging.StreamHandler()
console_handler.setLevel(lvl)
console_handler.setFormatter(format_str)
self.console_logger.addHandler(console_handler)
if lvl == logging.DEBUG:
filename = os.path.join(os.path.dirname(__file__), 'logs', filename)
self.logger_file_path = filename
# Automatically generates the logging file at specified intervals
file_handler = handlers.TimedRotatingFileHandler(
# If more than (backcount+1) existed, oldest logs will be deleted
filename=filename, when=when, backupCount=backcount, encoding='utf-8'
)
file_handler.setLevel(lvl)
file_handler.setFormatter(format_str)
# This for the logger when it appends the date to the new log
file_handler.namer = lambda name: name.replace(".log", "") + ".log"
self.console_logger.addHandler(file_handler)
self.debug(f'Debug log path: {self.logger_file_path}')
else:
self.logger_file_path = None
def debug(self, message: str) -> None:
self.console_logger.debug(message)
def info(self, message: str) -> None:
self.console_logger.info(message)
def warning(self, message: str) -> None:
self.console_logger.warning(message)
def error(self, message: str) -> None:
self.console_logger.error(message)
def critical(self, message: str) -> None:
self.console_logger.critical(message)
class _StorageTools:
@staticmethod
def __clean_attributes(element: html.HtmlElement, forbidden: tuple = ()) -> Dict:
+7
View File
@@ -0,0 +1,7 @@
from .camo import CamoufoxEngine
from .static import StaticEngine
from .pw import PlaywrightEngine
from .constants import DEFAULT_DISABLED_RESOURCES, DEFAULT_STEALTH_FLAGS
from .toolbelt import check_if_engine_usable
__all__ = ['CamoufoxEngine', 'PlaywrightEngine']
+121
View File
@@ -0,0 +1,121 @@
import logging
from scrapling.core._types import Union, Callable, Optional, Dict, List, Literal
from scrapling.engines.toolbelt import (
Response,
do_nothing,
get_os_name,
intercept_route,
check_type_validity,
generate_convincing_referer,
)
from camoufox.sync_api import Camoufox
class CamoufoxEngine:
def __init__(
self, headless: Optional[Union[bool, Literal['virtual']]] = True, block_images: Optional[bool] = False, disable_resources: Optional[bool] = False,
block_webrtc: Optional[bool] = False, allow_webgl: Optional[bool] = False, network_idle: Optional[bool] = False, humanize: Optional[Union[bool, float]] = True,
timeout: Optional[float] = 30000, page_action: Callable = do_nothing, wait_selector: Optional[str] = None, addons: Optional[List[str]] = None,
wait_selector_state: str = 'attached', google_search: Optional[bool] = True, extra_headers: Optional[Dict[str, str]] = None, adaptor_arguments: Dict = None
):
"""An engine that utilizes Camoufox library, check the `StealthyFetcher` class for more documentation.
:param headless: Run the browser in headless/hidden (default), virtual screen mode, or headful/visible mode.
:param block_images: Prevent the loading of images through Firefox preferences.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param disable_resources: Drop requests of unnecessary resources for a speed boost. It depends but it made requests ~25% faster in my tests for some websites.
Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param block_webrtc: Blocks WebRTC entirely.
:param addons: List of Firefox addons to use. Must be paths to extracted addons.
:param humanize: Humanize the cursor movement. Takes either True or the MAX duration in seconds of the cursor movement. The cursor typically takes up to 1.5 seconds to move across the window.
:param allow_webgl: Whether to allow WebGL. To prevent leaks, only use this for special cases.
:param network_idle: Wait for the page until there are no network connections for at least 500 ms.
:param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000
:param page_action: Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again.
:param wait_selector: Wait for a specific css selector to be in a specific state.
:param wait_selector_state: The state to wait for the selector given with `wait_selector`. Default state is `attached`.
:param google_search: Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name.
:param extra_headers: A dictionary of extra headers to add to the request. _The referer set by the `google_search` argument takes priority over the referer set here if used together._
:param adaptor_arguments: The arguments that will be passed in the end while creating the final Adaptor's class.
"""
self.headless = headless
self.block_images = bool(block_images)
self.disable_resources = bool(disable_resources)
self.block_webrtc = bool(block_webrtc)
self.allow_webgl = bool(allow_webgl)
self.network_idle = bool(network_idle)
self.google_search = bool(google_search)
self.extra_headers = extra_headers or {}
self.addons = addons or []
self.humanize = humanize
self.timeout = check_type_validity(timeout, [int, float], 30000)
if callable(page_action):
self.page_action = page_action
else:
self.page_action = do_nothing
logging.error('[Ignored] Argument "page_action" must be callable')
self.wait_selector = wait_selector
self.wait_selector_state = wait_selector_state
self.adaptor_arguments = adaptor_arguments if adaptor_arguments else {}
def fetch(self, url: str) -> Response:
"""Opens up the browser and do your request based on your chosen options.
:param url: Target url.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
with Camoufox(
headless=self.headless,
block_images=self.block_images, # Careful! it makes some websites doesn't finish loading at all like stackoverflow even in headful
os=get_os_name(),
block_webrtc=self.block_webrtc,
allow_webgl=self.allow_webgl,
addons=self.addons,
humanize=self.humanize,
i_know_what_im_doing=True, # To turn warnings off with user configurations
) as browser:
page = browser.new_page()
page.set_default_navigation_timeout(self.timeout)
page.set_default_timeout(self.timeout)
if self.disable_resources:
page.route("**/*", intercept_route)
if self.extra_headers:
page.set_extra_http_headers(self.extra_headers)
res = page.goto(url, referer=generate_convincing_referer(url) if self.google_search else None)
page.wait_for_load_state(state="domcontentloaded")
if self.network_idle:
page.wait_for_load_state('networkidle')
page = self.page_action(page)
if self.wait_selector and type(self.wait_selector) is str:
waiter = page.locator(self.wait_selector)
waiter.wait_for(state=self.wait_selector_state)
content_type = res.headers.get('content-type', '')
# Parse charset from content-type
encoding = 'utf-8' # default encoding
if 'charset=' in content_type.lower():
encoding = content_type.lower().split('charset=')[-1].split(';')[0].strip()
response = Response(
url=res.url,
text=page.content(),
content=res.body(),
status=res.status,
reason=res.status_text,
encoding=encoding,
cookies={cookie['name']: cookie['value'] for cookie in page.context.cookies()},
headers=res.all_headers(),
request_headers=res.request.all_headers(),
adaptor_arguments=self.adaptor_arguments
)
page.close()
return response
+108
View File
@@ -0,0 +1,108 @@
# Disable loading these resources for speed
DEFAULT_DISABLED_RESOURCES = [
'font',
'image',
'media',
'beacon',
'object',
'imageset',
'texttrack',
'websocket',
'csp_report',
'stylesheet',
]
DEFAULT_STEALTH_FLAGS = [
# Explanation: https://peter.sh/experiments/chromium-command-line-switches/
# Generally this will make the browser faster and less detectable
'--no-pings',
'--incognito',
'--test-type',
'--lang=en-US',
'--mute-audio',
'--no-first-run',
'--disable-sync',
'--hide-scrollbars',
'--disable-logging',
'--start-maximized', # For headless check bypass
'--enable-async-dns',
'--disable-breakpad',
'--disable-infobars',
'--accept-lang=en-US',
'--use-mock-keychain',
'--disable-translate',
'--disable-extensions',
'--disable-voice-input',
'--window-position=0,0',
'--disable-wake-on-wifi',
'--ignore-gpu-blocklist',
'--enable-tcp-fast-open',
'--enable-web-bluetooth',
'--disable-hang-monitor',
'--password-store=basic',
'--disable-cloud-import',
'--disable-default-apps',
'--disable-print-preview',
'--disable-dev-shm-usage',
'--disable-popup-blocking',
'--metrics-recording-only',
'--disable-crash-reporter',
'--disable-partial-raster',
'--disable-gesture-typing',
'--disable-checker-imaging',
'--disable-prompt-on-repost',
'--force-color-profile=srgb',
'--font-render-hinting=none',
'--no-default-browser-check',
'--aggressive-cache-discard',
'--disable-component-update',
'--disable-cookie-encryption',
'--disable-domain-reliability',
'--disable-threaded-animation',
'--disable-threaded-scrolling',
# '--disable-reading-from-canvas', # For Firefox
'--enable-simple-cache-backend',
'--disable-background-networking',
'--disable-session-crashed-bubble',
'--enable-surface-synchronization',
'--disable-image-animation-resync',
'--disable-renderer-backgrounding',
'--disable-ipc-flooding-protection',
'--prerender-from-omnibox=disabled',
'--safebrowsing-disable-auto-update',
'--disable-offer-upload-credit-cards',
'--disable-features=site-per-process',
'--disable-background-timer-throttling',
'--disable-new-content-rendering-timeout',
'--run-all-compositor-stages-before-draw',
'--disable-client-side-phishing-detection',
'--disable-backgrounding-occluded-windows',
'--disable-layer-tree-host-memory-pressure',
'--autoplay-policy=no-user-gesture-required',
'--disable-offer-store-unmasked-wallet-cards',
'--disable-blink-features=AutomationControlled',
'--webrtc-ip-handling-policy=disable_non_proxied_udp',
'--disable-component-extensions-with-background-pages',
'--force-webrtc-ip-handling-policy=disable_non_proxied_udp',
'--enable-features=NetworkService,NetworkServiceInProcess,TrustTokens,TrustTokensAlwaysAllowIssuance',
'--blink-settings=primaryHoverType=2,availableHoverTypes=2,primaryPointerType=4,availablePointerTypes=4',
'--disable-features=AudioServiceOutOfProcess,IsolateOrigins,site-per-process,TranslateUI,BlinkGenPropertyTrees',
]
# Defaulting to the docker mode, token doesn't matter in it as it's passed for the container
NSTBROWSER_DEFAULT_QUERY = {
"once": True,
"headless": True,
"autoClose": True,
"fingerprint": {
"flags": {
"timezone": "BasedOnIp",
"screen": "Custom"
},
"platform": 'linux', # support: windows, mac, linux
"kernel": 'chromium', # only support: chromium
"kernelMilestone": '128',
"hardwareConcurrency": 8,
"deviceMemory": 8,
},
}
+232
View File
@@ -0,0 +1,232 @@
import json
import logging
from scrapling.core._types import Union, Callable, Optional, List, Dict
from scrapling.engines.constants import DEFAULT_STEALTH_FLAGS, NSTBROWSER_DEFAULT_QUERY
from scrapling.engines.toolbelt import (
Response,
do_nothing,
js_bypass_path,
intercept_route,
generate_headers,
check_type_validity,
construct_cdp_url,
generate_convincing_referer,
)
class PlaywrightEngine:
def __init__(
self, headless: Union[bool, str] = True,
disable_resources: bool = False,
useragent: Optional[str] = None,
network_idle: Optional[bool] = False,
timeout: Optional[float] = 30000,
page_action: Callable = do_nothing,
wait_selector: Optional[str] = None,
wait_selector_state: Optional[str] = 'attached',
stealth: bool = False,
hide_canvas: bool = True,
disable_webgl: bool = False,
cdp_url: Optional[str] = None,
nstbrowser_mode: bool = False,
nstbrowser_config: Optional[Dict] = None,
google_search: Optional[bool] = True,
extra_headers: Optional[Dict[str, str]] = None,
adaptor_arguments: Dict = None
):
"""An engine that utilizes PlayWright library, check the `PlayWrightFetcher` class for more documentation.
:param headless: Run the browser in headless/hidden (default), or headful/visible mode.
:param disable_resources: Drop requests of unnecessary resources for a speed boost. It depends but it made requests ~25% faster in my tests for some websites.
Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param useragent: Pass a useragent string to be used. Otherwise the fetcher will generate a real Useragent of the same browser and use it.
:param network_idle: Wait for the page until there are no network connections for at least 500 ms.
:param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000
:param page_action: Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again.
:param wait_selector: Wait for a specific css selector to be in a specific state.
:param wait_selector_state: The state to wait for the selector given with `wait_selector`. Default state is `attached`.
:param stealth: Enables stealth mode, check the documentation to see what stealth mode does currently.
:param hide_canvas: Add random noise to canvas operations to prevent fingerprinting.
:param disable_webgl: Disables WebGL and WebGL 2.0 support entirely.
:param cdp_url: Instead of launching a new browser instance, connect to this CDP URL to control real browsers/NSTBrowser through CDP.
:param nstbrowser_mode: Enables NSTBrowser mode, it have to be used with `cdp_url` argument or it will get completely ignored.
:param google_search: Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name.
:param extra_headers: A dictionary of extra headers to add to the request. _The referer set by the `google_search` argument takes priority over the referer set here if used together._
:param nstbrowser_config: The config you want to send with requests to the NSTBrowser. If left empty, Scrapling defaults to an optimized NSTBrowser's docker browserless config.
:param adaptor_arguments: The arguments that will be passed in the end while creating the final Adaptor's class.
"""
self.headless = headless
self.disable_resources = disable_resources
self.network_idle = bool(network_idle)
self.stealth = bool(stealth)
self.hide_canvas = bool(hide_canvas)
self.disable_webgl = bool(disable_webgl)
self.google_search = bool(google_search)
self.extra_headers = extra_headers or {}
self.cdp_url = cdp_url
self.useragent = useragent
self.timeout = check_type_validity(timeout, [int, float], 30000)
if callable(page_action):
self.page_action = page_action
else:
self.page_action = do_nothing
logging.error('[Ignored] Argument "page_action" must be callable')
self.wait_selector = wait_selector
self.wait_selector_state = wait_selector_state
self.nstbrowser_mode = bool(nstbrowser_mode)
self.nstbrowser_config = nstbrowser_config
self.adaptor_arguments = adaptor_arguments if adaptor_arguments else {}
def _cdp_url_logic(self, flags: Optional[List] = None) -> str:
"""Constructs new CDP URL if NSTBrowser is enabled otherwise return CDP URL as it is
:param flags: Chrome flags to be added to NSTBrowser query
:return: CDP URL
"""
cdp_url = self.cdp_url
if self.nstbrowser_mode:
if self.nstbrowser_config and type(self.nstbrowser_config) is Dict:
config = self.nstbrowser_config
else:
query = NSTBROWSER_DEFAULT_QUERY.copy()
if flags:
query.update({
"args": dict(zip(flags, [''] * len(flags))), # browser args should be a dictionary
})
config = {
'config': json.dumps(query),
# 'token': ''
}
cdp_url = construct_cdp_url(cdp_url, config)
else:
# To validate it
cdp_url = construct_cdp_url(cdp_url)
return cdp_url
def fetch(self, url: str) -> Response:
"""Opens up the browser and do your request based on your chosen options.
:param url: Target url.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
if not self.stealth:
from playwright.sync_api import sync_playwright
else:
from rebrowser_playwright.sync_api import sync_playwright
with sync_playwright() as p:
# Handle the UserAgent early
if self.useragent:
extra_headers = {}
useragent = self.useragent
else:
extra_headers = generate_headers(browser_mode=True)
useragent = extra_headers.get('User-Agent')
# Prepare the flags before diving
flags = DEFAULT_STEALTH_FLAGS
if self.hide_canvas:
flags += ['--fingerprinting-canvas-image-data-noise']
if self.disable_webgl:
flags += ['--disable-webgl', '--disable-webgl-image-chromium', '--disable-webgl2']
# Creating the browser
if self.cdp_url:
cdp_url = self._cdp_url_logic(flags if self.stealth else None)
browser = p.chromium.connect_over_cdp(endpoint_url=cdp_url)
else:
if self.stealth:
browser = p.chromium.launch(headless=self.headless, args=flags, ignore_default_args=['--enable-automation'], chromium_sandbox=True)
else:
browser = p.chromium.launch(headless=self.headless, ignore_default_args=['--enable-automation'])
# Creating the context
if self.stealth:
context = browser.new_context(
locale='en-US',
is_mobile=False,
has_touch=False,
color_scheme='dark', # Bypasses the 'prefersLightColor' check in creepjs
user_agent=useragent,
device_scale_factor=2,
# I'm thinking about disabling it to rest from all Service Workers headache but let's keep it as it is for now
service_workers="allow",
ignore_https_errors=True,
extra_http_headers=extra_headers,
screen={"width": 1920, "height": 1080},
viewport={"width": 1920, "height": 1080},
permissions=["geolocation", 'notifications'],
)
else:
context = browser.new_context(
color_scheme='dark',
user_agent=useragent,
device_scale_factor=2,
extra_http_headers=extra_headers
)
# Finally we are in business
page = context.new_page()
page.set_default_navigation_timeout(self.timeout)
page.set_default_timeout(self.timeout)
if self.extra_headers:
page.set_extra_http_headers(self.extra_headers)
if self.disable_resources:
page.route("**/*", intercept_route)
if self.stealth:
# Basic bypasses nothing fancy as I'm still working on it
# But with adding these bypasses to the above config, it bypasses many online tests like
# https://bot.sannysoft.com/
# https://kaliiiiiiiiii.github.io/brotector/
# https://pixelscan.net/
# https://iphey.com/
# https://www.browserscan.net/bot-detection <== this one also checks for the CDP runtime fingerprint
# https://arh.antoinevastel.com/bots/areyouheadless/
# https://prescience-data.github.io/execution-monitor.html
page.add_init_script(path=js_bypass_path('webdriver_fully.js'))
page.add_init_script(path=js_bypass_path('window_chrome.js'))
page.add_init_script(path=js_bypass_path('navigator_plugins.js'))
page.add_init_script(path=js_bypass_path('pdf_viewer.js'))
page.add_init_script(path=js_bypass_path('notification_permission.js'))
page.add_init_script(path=js_bypass_path('screen_props.js'))
page.add_init_script(path=js_bypass_path('playwright_fingerprint.js'))
res = page.goto(url, referer=generate_convincing_referer(url) if self.google_search else None)
page.wait_for_load_state(state="domcontentloaded")
if self.network_idle:
page.wait_for_load_state('networkidle')
page = self.page_action(page)
if self.wait_selector and type(self.wait_selector) is str:
waiter = page.locator(self.wait_selector)
waiter.wait_for(state=self.wait_selector_state)
content_type = res.headers.get('content-type', '')
# Parse charset from content-type
encoding = 'utf-8' # default encoding
if 'charset=' in content_type.lower():
encoding = content_type.lower().split('charset=')[-1].split(';')[0].strip()
response = Response(
url=res.url,
text=page.content(),
content=res.body(),
status=res.status,
reason=res.status_text,
encoding=encoding,
cookies={cookie['name']: cookie['value'] for cookie in page.context.cookies()},
headers=res.all_headers(),
request_headers=res.request.all_headers(),
adaptor_arguments=self.adaptor_arguments
)
page.close()
return response
+112
View File
@@ -0,0 +1,112 @@
import logging
from scrapling.core._types import Union, Optional, Dict
from .toolbelt import Response, generate_convincing_referer, generate_headers
import httpx
from httpx._models import Response as httpxResponse
class StaticEngine:
def __init__(self, follow_redirects: bool = True, timeout: Optional[Union[int, float]] = None, adaptor_arguments: Dict = None):
"""An engine that utilizes httpx library, check the `Fetcher` class for more documentation.
:param follow_redirects: As the name says -- if enabled (default), redirects will be followed.
:param timeout: The time to wait for the request to finish in seconds. The default is 10 seconds.
:param adaptor_arguments: The arguments that will be passed in the end while creating the final Adaptor's class.
"""
self.timeout = timeout
self.follow_redirects = bool(follow_redirects)
self._extra_headers = generate_headers(browser_mode=False)
self.adaptor_arguments = adaptor_arguments if adaptor_arguments else {}
@staticmethod
def _headers_job(headers: Optional[Dict], url: str, stealth: bool) -> Dict:
"""Adds useragent to headers if it doesn't exist, generates real headers and append it to current headers, and
finally generates a referer header that looks like if this request came from Google's search of the current URL's domain.
:param headers: Current headers in the request if the user passed any
:param url: The Target URL.
:param stealth: Whether stealth mode is enabled or not.
:return: A dictionary of the new headers.
"""
headers = headers or {}
# Validate headers
if not headers.get('user-agent') and not headers.get('User-Agent'):
headers['User-Agent'] = generate_headers(browser_mode=False).get('User-Agent')
logging.info(f"Can't find useragent in headers so '{headers['User-Agent']}' was used.")
if stealth:
extra_headers = generate_headers(browser_mode=False)
headers.update(extra_headers)
headers.update({'referer': generate_convincing_referer(url)})
return headers
def _prepare_response(self, response: httpxResponse) -> Response:
"""Takes httpx response and generates `Response` object from it.
:param response: httpx response object
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
return Response(
url=str(response.url),
text=response.text,
content=response.content,
status=response.status_code,
reason=response.reason_phrase,
encoding=response.encoding or 'utf-8',
cookies=dict(response.cookies),
headers=dict(response.headers),
request_headers=dict(response.request.headers),
adaptor_arguments=self.adaptor_arguments
)
def get(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP GET request for you but with some added flavors.
:param url: Target url.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request had came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.get()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
request = httpx.get(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
return self._prepare_response(request)
def post(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP POST request for you but with some added flavors.
:param url: Target url.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request had came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.post()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
request = httpx.post(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
return self._prepare_response(request)
def delete(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP DELETE request for you but with some added flavors.
:param url: Target url.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request had came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.delete()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
request = httpx.delete(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
return self._prepare_response(request)
def put(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP PUT request for you but with some added flavors.
:param url: Target url.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request had came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.put()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
request = httpx.put(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
return self._prepare_response(request)
+18
View File
@@ -0,0 +1,18 @@
from .fingerprints import (
get_os_name,
generate_headers,
generate_convincing_referer,
)
from .custom import (
Response,
do_nothing,
BaseFetcher,
get_variable_name,
check_type_validity,
check_if_engine_usable,
)
from .navigation import (
js_bypass_path,
intercept_route,
construct_cdp_url,
)
@@ -0,0 +1,40 @@
if(navigator.plugins.length == 0){
Object.defineProperty(navigator, 'plugins', {
get: () => {
const PDFViewerPlugin = Object.create(Plugin.prototype, {
description: { value: 'Portable Document Format', enumerable: false },
filename: { value: 'internal-pdf-viewer', enumerable: false },
name: { value: 'PDF Viewer', enumerable: false },
});
const ChromePDFViewer = Object.create(Plugin.prototype, {
description: { value: 'Portable Document Format', enumerable: false },
filename: { value: 'internal-pdf-viewer', enumerable: false },
name: { value: 'Chrome PDF Viewer', enumerable: false },
});
const ChromiumPDFViewer = Object.create(Plugin.prototype, {
description: { value: 'Portable Document Format', enumerable: false },
filename: { value: 'internal-pdf-viewer', enumerable: false },
name: { value: 'Chromium PDF Viewer', enumerable: false },
});
const EdgePDFViewer = Object.create(Plugin.prototype, {
description: { value: 'Portable Document Format', enumerable: false },
filename: { value: 'internal-pdf-viewer', enumerable: false },
name: { value: 'Microsoft Edge PDF Viewer', enumerable: false },
});
const WebKitPDFPlugin = Object.create(Plugin.prototype, {
description: { value: 'Portable Document Format', enumerable: false },
filename: { value: 'internal-pdf-viewer', enumerable: false },
name: { value: 'WebKit built-in PDF', enumerable: false },
});
return Object.create(PluginArray.prototype, {
length: { value: 5 },
0: { value: PDFViewerPlugin },
1: { value: ChromePDFViewer },
2: { value: ChromiumPDFViewer },
3: { value: EdgePDFViewer },
4: { value: WebKitPDFPlugin },
});
},
});
}
@@ -0,0 +1,5 @@
// Bypasses `notificationIsDenied` test in creepsjs's 'Like Headless' sections
const isSecure = document.location.protocol.startsWith('https')
if (isSecure){
Object.defineProperty(Notification, 'permission', {get: () => 'default'})
}
@@ -0,0 +1,5 @@
// PDF viewer enabled
// Bypasses `pdfIsDisabled` test in creepsjs's 'Like Headless' sections
Object.defineProperty(navigator, 'pdfViewerEnabled', {
get: () => true,
});
@@ -0,0 +1,2 @@
// Remove playwright fingerprint => https://github.com/microsoft/playwright/commit/c9e673c6dca746384338ab6bb0cf63c7e7caa9b2#diff-087773eea292da9db5a3f27de8f1a2940cdb895383ad750c3cd8e01772a35b40R915
delete __pwInitScripts;
@@ -0,0 +1,27 @@
const windowScreenProps = {
// Dimensions
innerHeight: 0,
innerWidth: 0,
outerHeight: 754,
outerWidth: 1313,
// Position
screenX: 19,
pageXOffset: 0,
pageYOffset: 0,
// Display
devicePixelRatio: 2
};
try {
for (const [prop, value] of Object.entries(windowScreenProps)) {
if (value > 0) {
// The 0 values are introduced by collecting in the hidden iframe.
// They are document sizes anyway so no need to test them or inject them.
window[prop] = value;
}
}
} catch (e) {
console.warn(e);
};
@@ -0,0 +1,27 @@
// Create a function that looks like a native getter
const nativeGetter = function get webdriver() {
return false;
};
// Copy over native function properties
Object.defineProperties(nativeGetter, {
name: { value: 'get webdriver', configurable: true },
length: { value: 0, configurable: true },
toString: {
value: function() {
return `function get webdriver() { [native code] }`;
},
configurable: true
}
});
// Make it look native
Object.setPrototypeOf(nativeGetter, Function.prototype);
// Apply the modified descriptor
Object.defineProperty(Navigator.prototype, 'webdriver', {
get: nativeGetter,
set: undefined,
enumerable: true,
configurable: true
});
@@ -0,0 +1,213 @@
// To escape `HEADCHR_CHROME_OBJ` test in headless mode => https://github.com/antoinevastel/fp-collect/blob/master/src/fpCollect.js#L322
// Faking window.chrome fully
if (!window.chrome) {
// First, save all existing properties
const originalKeys = Object.getOwnPropertyNames(window);
const tempObj = {};
// Recreate all properties in original order
for (const key of originalKeys) {
const descriptor = Object.getOwnPropertyDescriptor(window, key);
const value = window[key];
// delete window[key];
Object.defineProperty(tempObj, key, descriptor);
}
// Use the exact property descriptor found in headful Chrome
// fetch it via `Object.getOwnPropertyDescriptor(window, 'chrome')`
const mockChrome = {
loadTimes: {},
csi: {},
app: {
isInstalled: false
},
// Add other Chrome-specific properties
};
Object.defineProperty(tempObj, 'chrome', {
writable: true,
enumerable: true,
configurable: false,
value: mockChrome
});
for (const key of Object.getOwnPropertyNames(tempObj)) {
try {
Object.defineProperty(window, key,
Object.getOwnPropertyDescriptor(tempObj, key));
} catch (e) {}
};
// todo: solve this
// Using line below bypasses the hasHighChromeIndex test in creepjs ==> https://github.com/abrahamjuliot/creepjs/blob/master/src/headless/index.ts#L121
// Chrome object have to be in the end of the window properties
// Object.assign(window, tempObj);
// But makes window.chrome unreadable on 'https://bot.sannysoft.com/'
}
// That means we're running headful and don't need to mock anything
if ('app' in window.chrome) {
return; // Nothing to do here
}
const makeError = {
ErrorInInvocation: fn => {
const err = new TypeError(`Error in invocation of app.${fn}()`);
return utils.stripErrorWithAnchor(
err,
`at ${fn} (eval at <anonymous>`,
);
},
};
// check with: `JSON.stringify(window.chrome['app'])`
const STATIC_DATA = JSON.parse(
`
{
"isInstalled": false,
"InstallState": {
"DISABLED": "disabled",
"INSTALLED": "installed",
"NOT_INSTALLED": "not_installed"
},
"RunningState": {
"CANNOT_RUN": "cannot_run",
"READY_TO_RUN": "ready_to_run",
"RUNNING": "running"
}
}
`.trim(),
);
window.chrome.app = {
...STATIC_DATA,
get isInstalled() {
return false;
},
getDetails: function getDetails() {
if (arguments.length) {
throw makeError.ErrorInInvocation(`getDetails`);
}
return null;
},
getIsInstalled: function getDetails() {
if (arguments.length) {
throw makeError.ErrorInInvocation(`getIsInstalled`);
}
return false;
},
runningState: function getDetails() {
if (arguments.length) {
throw makeError.ErrorInInvocation(`runningState`);
}
return 'cannot_run';
},
};
// Check that the Navigation Timing API v1 is available, we need that
if (!window.performance || !window.performance.timing) {
return;
}
const {timing} = window.performance;
window.chrome.csi = function () {
return {
onloadT: timing.domContentLoadedEventEnd,
startE: timing.navigationStart,
pageT: Date.now() - timing.navigationStart,
tran: 15, // Transition type or something
};
};
if (!window.PerformancePaintTiming){
return;
}
const {performance} = window;
// Some stuff is not available on about:blank as it requires a navigation to occur,
// let's harden the code to not fail then:
const ntEntryFallback = {
nextHopProtocol: 'h2',
type: 'other',
};
// The API exposes some funky info regarding the connection
const protocolInfo = {
get connectionInfo() {
const ntEntry =
performance.getEntriesByType('navigation')[0] || ntEntryFallback;
return ntEntry.nextHopProtocol;
},
get npnNegotiatedProtocol() {
// NPN is deprecated in favor of ALPN, but this implementation returns the
// HTTP/2 or HTTP2+QUIC/39 requests negotiated via ALPN.
const ntEntry =
performance.getEntriesByType('navigation')[0] || ntEntryFallback;
return ['h2', 'hq'].includes(ntEntry.nextHopProtocol)
? ntEntry.nextHopProtocol
: 'unknown';
},
get navigationType() {
const ntEntry =
performance.getEntriesByType('navigation')[0] || ntEntryFallback;
return ntEntry.type;
},
get wasAlternateProtocolAvailable() {
// The Alternate-Protocol header is deprecated in favor of Alt-Svc
// (https://www.mnot.net/blog/2016/03/09/alt-svc), so technically this
// should always return false.
return false;
},
get wasFetchedViaSpdy() {
// SPDY is deprecated in favor of HTTP/2, but this implementation returns
// true for HTTP/2 or HTTP2+QUIC/39 as well.
const ntEntry =
performance.getEntriesByType('navigation')[0] || ntEntryFallback;
return ['h2', 'hq'].includes(ntEntry.nextHopProtocol);
},
get wasNpnNegotiated() {
// NPN is deprecated in favor of ALPN, but this implementation returns true
// for HTTP/2 or HTTP2+QUIC/39 requests negotiated via ALPN.
const ntEntry =
performance.getEntriesByType('navigation')[0] || ntEntryFallback;
return ['h2', 'hq'].includes(ntEntry.nextHopProtocol);
},
};
// Truncate number to specific number of decimals, most of the `loadTimes` stuff has 3
function toFixed(num, fixed) {
var re = new RegExp('^-?\\d+(?:.\\d{0,' + (fixed || -1) + '})?');
return num.toString().match(re)[0];
}
const timingInfo = {
get firstPaintAfterLoadTime() {
// This was never actually implemented and always returns 0.
return 0;
},
get requestTime() {
return timing.navigationStart / 1000;
},
get startLoadTime() {
return timing.navigationStart / 1000;
},
get commitLoadTime() {
return timing.responseStart / 1000;
},
get finishDocumentLoadTime() {
return timing.domContentLoadedEventEnd / 1000;
},
get finishLoadTime() {
return timing.loadEventEnd / 1000;
},
get firstPaintTime() {
const fpEntry = performance.getEntriesByType('paint')[0] || {
startTime: timing.loadEventEnd / 1000, // Fallback if no navigation occured (`about:blank`)
};
return toFixed(
(fpEntry.startTime + performance.timeOrigin) / 1000,
3,
);
},
};
window.chrome.loadTimes = function () {
return {
...protocolInfo,
...timingInfo,
};
};
+168
View File
@@ -0,0 +1,168 @@
"""
Functions related to custom types or type checking
"""
import inspect
import logging
from dataclasses import dataclass, field
from scrapling.core.utils import setup_basic_logging
from scrapling.parser import Adaptor, SQLiteStorageSystem
from scrapling.core._types import Any, List, Type, Union, Optional, Dict, Callable
@dataclass(frozen=True)
class Response:
"""This class is returned by all engines as a way to unify response type between different libraries."""
url: str
text: str
content: bytes
status: int
reason: str
encoding: str = 'utf-8' # default encoding
cookies: Dict = field(default_factory=dict)
headers: Dict = field(default_factory=dict)
request_headers: Dict = field(default_factory=dict)
adaptor_arguments: Dict = field(default_factory=dict)
@property
def adaptor(self) -> Union[Adaptor, None]:
"""Generate Adaptor instance from this response if possible, otherwise return None"""
automatch_domain = self.adaptor_arguments.pop('automatch_domain', None)
if self.text:
# For playwright that will be the response after all JS executed
return Adaptor(text=self.text, url=automatch_domain or self.url, encoding=self.encoding, **self.adaptor_arguments)
elif self.content:
# For playwright, that's after all JS is loaded but not all of them executed, because playwright doesn't offer something like page.content()
# To get response Bytes after the load states
# Reference: https://playwright.dev/python/docs/api/class-page
return Adaptor(body=self.content, url=automatch_domain or self.url, encoding=self.encoding, **self.adaptor_arguments)
return None
def __repr__(self):
return f'<{self.__class__.__name__} [{self.status} {self.reason}]>'
class BaseFetcher:
def __init__(
self, huge_tree: bool = True, keep_comments: Optional[bool] = False, auto_match: Optional[bool] = True,
storage: Any = SQLiteStorageSystem, storage_args: Optional[Dict] = None, debug: Optional[bool] = True,
automatch_domain: Optional[str] = None,
):
"""Arguments below are the same from the Adaptor class so you can pass them directly, the rest of Adaptor's arguments
are detected and passed automatically from the Fetcher based on the response for accessibility.
:param huge_tree: Enabled by default, should always be enabled when parsing large HTML documents. This controls
libxml2 feature that forbids parsing certain large documents to protect from possible memory exhaustion.
:param keep_comments: While parsing the HTML body, drop comments or not. Disabled by default for obvious reasons
:param auto_match: Globally turn-off the auto-match feature in all functions, this argument takes higher
priority over all auto-match related arguments/functions in the class.
:param storage: The storage class to be passed for auto-matching functionalities, see ``Docs`` for more info.
:param storage_args: A dictionary of ``argument->value`` pairs to be passed for the storage class.
If empty, default values will be used.
:param automatch_domain: For cases where you want to automatch selectors across different websites as if they were on the same website, use this argument to unify them.
Otherwise, the domain of the request is used by default.
:param debug: Enable debug mode
"""
# Adaptor class parameters
# I won't validate Adaptor's class parameters here again, I will leave it to be validated later
self.adaptor_arguments = dict(
huge_tree=huge_tree,
keep_comments=keep_comments,
auto_match=auto_match,
storage=storage,
storage_args=storage_args,
debug=debug,
)
# If the user used fetchers first, then configure the logger from here instead of the `Adaptor` class
setup_basic_logging(level='debug' if debug else 'info')
if automatch_domain:
if type(automatch_domain) is not str:
logging.warning('[Ignored] The argument "automatch_domain" must be of string type')
else:
self.adaptor_arguments.update({'automatch_domain': automatch_domain})
def check_if_engine_usable(engine: Callable) -> Union[Callable, None]:
"""This function check if the passed engine can be used by a Fetcher-type class or not.
:param engine: The engine class itself
:return: The engine class again if all checks out, otherwise raises error
:raise TypeError: If engine class don't have fetch method, If engine class have fetch attribute not method, or If engine class have fetch function but it doesn't take arguments
"""
# if isinstance(engine, type):
# raise TypeError("Expected an engine instance, not a class definition of the engine")
if hasattr(engine, 'fetch'):
fetch_function = getattr(engine, "fetch")
if callable(fetch_function):
if len(inspect.signature(fetch_function).parameters) > 0:
return engine
else:
# raise TypeError("Engine class instance must have a callable method 'fetch' with the first argument used for the url.")
raise TypeError("Engine class must have a callable method 'fetch' with the first argument used for the url.")
else:
# raise TypeError("Invalid engine instance! Engine class must have a callable method 'fetch'")
raise TypeError("Invalid engine class! Engine class must have a callable method 'fetch'")
else:
# raise TypeError("Invalid engine instance! Engine class must have the method 'fetch'")
raise TypeError("Invalid engine class! Engine class must have the method 'fetch'")
def get_variable_name(var: Any) -> Optional[str]:
"""Get the name of a variable using global and local scopes.
:param var: The variable to find the name for
:return: The name of the variable if found, None otherwise
"""
for scope in [globals(), locals()]:
for name, value in scope.items():
if value is var:
return name
return None
def check_type_validity(variable: Any, valid_types: Union[List[Type], None], default_value: Any = None, critical: bool = False, param_name: Optional[str] = None) -> Any:
"""Check if a variable matches the specified type constraints.
:param variable: The variable to check
:param valid_types: List of valid types for the variable
:param default_value: Value to return if type check fails
:param critical: If True, raises TypeError instead of logging error
:param param_name: Optional parameter name for error messages
:return: The original variable if valid, default_value if invalid
:raise TypeError: If critical=True and type check fails
"""
# Use provided param_name or try to get it automatically
var_name = param_name or get_variable_name(variable) or "Unknown"
# Convert valid_types to a list if None
valid_types = valid_types or []
# Handle None value
if variable is None:
if type(None) in valid_types:
return variable
error_msg = f'Argument "{var_name}" cannot be None'
if critical:
raise TypeError(error_msg)
logging.error(f'[Ignored] {error_msg}')
return default_value
# If no valid_types specified and variable has a value, return it
if not valid_types:
return variable
# Check if variable type matches any of the valid types
if not any(isinstance(variable, t) for t in valid_types):
type_names = [t.__name__ for t in valid_types]
error_msg = f'Argument "{var_name}" must be of type {" or ".join(type_names)}'
if critical:
raise TypeError(error_msg)
logging.error(f'[Ignored] {error_msg}')
return default_value
return variable
# Pew Pew
def do_nothing(page):
# Just works as a filler for `page_action` argument in browser engines
return page
@@ -0,0 +1,81 @@
"""
Functions related to generating headers and fingerprints generally
"""
import platform
from scrapling.core.utils import cache
from scrapling.core._types import Union, Dict
from tldextract import extract
from browserforge.headers import HeaderGenerator, Browser
from browserforge.fingerprints import FingerprintGenerator, Fingerprint
@cache(None, typed=True)
def generate_convincing_referer(url: str) -> str:
"""Takes the domain from the URL without the subdomain/suffix and make it look like you were searching google for this website
>>> generate_convincing_referer('https://www.somewebsite.com/blah')
'https://www.google.com/search?q=somewebsite'
:param url: The URL you are about to fetch.
:return: Google's search URL of the domain name
"""
website_name = extract(url).domain
return f'https://www.google.com/search?q={website_name}'
@cache(None, typed=True)
def get_os_name() -> Union[str, None]:
"""Get the current OS name in the same format needed for browserforge
:return: Current OS name or `None` otherwise
"""
#
os_name = platform.system()
return {
'Linux': 'linux',
'Darwin': 'macos',
'Windows': 'windows',
# For the future? because why not
'iOS': 'ios',
}.get(os_name)
def generate_suitable_fingerprint() -> Fingerprint:
"""Generates a browserforge's fingerprint that matches current OS, desktop device, and Chrome with version 128 at least.
This function was originally created to test Browserforge's injector.
:return: `Fingerprint` object
"""
return FingerprintGenerator(
browser=[Browser(name='chrome', min_version=128)],
os=get_os_name(), # None is ignored
device='desktop'
).generate()
def generate_headers(browser_mode: bool = False) -> Dict:
"""Generate real browser-like headers using browserforge's generator
:param browser_mode: If enabled, the headers created are used for playwright so it have to match everything
:return: A dictionary of the generated headers
"""
if browser_mode:
# In this mode we don't care about anything other than matching the OS and the browser type with the browser we are using
# So we don't raise any inconsistency red flags while websites fingerprinting us
os_name = get_os_name()
return HeaderGenerator(
browser=[Browser(name='chrome', min_version=128)],
os=os_name, # None is ignored
device='desktop'
).generate()
else:
# Here it's used for normal requests that aren't done through browsers so we can take it lightly
browsers = [
Browser(name='chrome', min_version=120),
Browser(name='firefox', min_version=120),
Browser(name='edge', min_version=120),
]
return HeaderGenerator(browser=browsers, device='desktop').generate()
+74
View File
@@ -0,0 +1,74 @@
"""
Functions related to files and URLs
"""
import os
import logging
from urllib.parse import urlparse, urlencode
from scrapling.core.utils import cache
from scrapling.core._types import Union, Dict, Optional
from scrapling.engines.constants import DEFAULT_DISABLED_RESOURCES
from playwright.sync_api import Route
def intercept_route(route: Route) -> Union[Route, None]:
"""This is just a route handler but it drops requests that its type falls in `DEFAULT_DISABLED_RESOURCES`
:param route: PlayWright `Route` object of the current page
:return: PlayWright `Route` object
"""
if route.request.resource_type in DEFAULT_DISABLED_RESOURCES:
logging.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"')
return route.abort()
return route.continue_()
def construct_cdp_url(cdp_url: str, query_params: Optional[Dict] = None) -> str:
"""Takes a CDP URL, reconstruct it to check it's valid, then adds encoded parameters if exists
:param cdp_url: The target URL.
:param query_params: A dictionary of the parameters to add.
:return: The new CDP URL.
"""
try:
# Validate the base URL structure
parsed = urlparse(cdp_url)
# Check scheme
if parsed.scheme not in ('ws', 'wss'):
raise ValueError("CDP URL must use 'ws://' or 'wss://' scheme")
# Validate hostname and port
if not parsed.netloc:
raise ValueError("Invalid hostname for the CDP URL")
# Ensure path starts with /
path = parsed.path
if not path.startswith('/'):
path = '/' + path
# Reconstruct the base URL with validated parts
validated_base = f"{parsed.scheme}://{parsed.netloc}{path}"
# Add query parameters
if query_params:
query_string = urlencode(query_params)
return f"{validated_base}?{query_string}"
return validated_base
except Exception as e:
raise ValueError(f"Invalid CDP URL: {str(e)}")
@cache(None, typed=True)
def js_bypass_path(filename: str) -> str:
"""Takes the base filename of JS file inside the `bypasses` folder then return the full path of it
:param filename: The base filename of the JS file.
:return: The full path of the JS file.
"""
current_directory = os.path.dirname(__file__)
return os.path.join(current_directory, 'bypasses', filename)
+190
View File
@@ -0,0 +1,190 @@
from scrapling.core._types import Dict, Optional, Union, Callable, List, Literal
from scrapling.engines.toolbelt import Response, BaseFetcher, do_nothing
from scrapling.engines import CamoufoxEngine, PlaywrightEngine, StaticEngine, check_if_engine_usable
class Fetcher(BaseFetcher):
"""A basic `Fetcher` class type that can only do basic GET, POST, PUT, and DELETE HTTP requests based on httpx.
Any additional keyword arguments passed to the methods below are passed to the respective httpx's method directly.
"""
def get(self, url: str, follow_redirects: bool = True, timeout: Optional[Union[int, float]] = 10, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP GET request for you but with some added flavors.
:param url: Target url.
:param follow_redirects: As the name says -- if enabled (default), redirects will be followed.
:param timeout: The time to wait for the request to finish in seconds. The default is 10 seconds.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request had came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.get()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
response_object = StaticEngine(follow_redirects, timeout, adaptor_arguments=self.adaptor_arguments).get(url, stealthy_headers, **kwargs)
return response_object
def post(self, url: str, follow_redirects: bool = True, timeout: Optional[Union[int, float]] = 10, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP POST request for you but with some added flavors.
:param url: Target url.
:param follow_redirects: As the name says -- if enabled (default), redirects will be followed.
:param timeout: The time to wait for the request to finish in seconds. The default is 10 seconds.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.post()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
response_object = StaticEngine(follow_redirects, timeout, adaptor_arguments=self.adaptor_arguments).post(url, stealthy_headers, **kwargs)
return response_object
def put(self, url: str, follow_redirects: bool = True, timeout: Optional[Union[int, float]] = 10, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP PUT request for you but with some added flavors.
:param url: Target url
:param follow_redirects: As the name says -- if enabled (default), redirects will be followed.
:param timeout: The time to wait for the request to finish in seconds. The default is 10 seconds.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.put()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
response_object = StaticEngine(follow_redirects, timeout, adaptor_arguments=self.adaptor_arguments).put(url, stealthy_headers, **kwargs)
return response_object
def delete(self, url: str, follow_redirects: bool = True, timeout: Optional[Union[int, float]] = 10, stealthy_headers: Optional[bool] = True, **kwargs: Dict) -> Response:
"""Make basic HTTP DELETE request for you but with some added flavors.
:param url: Target url
:param follow_redirects: As the name says -- if enabled (default), redirects will be followed.
:param timeout: The time to wait for the request to finish in seconds. The default is 10 seconds.
:param stealthy_headers: If enabled (default), Fetcher will create and add real browser's headers and
create a referer header as if this request came from Google's search of this URL's domain.
:param kwargs: Any additional keyword arguments are passed directly to `httpx.delete()` function so check httpx documentation for details.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
response_object = StaticEngine(follow_redirects, timeout, adaptor_arguments=self.adaptor_arguments).delete(url, stealthy_headers, **kwargs)
return response_object
class StealthyFetcher(BaseFetcher):
"""A `Fetcher` class type that is completely stealthy fetcher that uses a modified version of Firefox.
It works as real browsers passing almost all online tests/protections based on Camoufox.
Other added flavors include setting the faked OS fingerprints to match the user's OS and the referer of every request is set as if this request came from Google's search of this URL's domain.
"""
def fetch(
self, url: str, headless: Optional[Union[bool, Literal['virtual']]] = True, block_images: Optional[bool] = False, disable_resources: Optional[bool] = False,
block_webrtc: Optional[bool] = False, allow_webgl: Optional[bool] = False, network_idle: Optional[bool] = False, addons: Optional[List[str]] = None,
timeout: Optional[float] = 30000, page_action: Callable = do_nothing, wait_selector: Optional[str] = None, humanize: Optional[Union[bool, float]] = True,
wait_selector_state: str = 'attached', google_search: Optional[bool] = True, extra_headers: Optional[Dict[str, str]] = None
) -> Response:
"""
Opens up a browser and do your request based on your chosen options below.
:param url: Target url.
:param headless: Run the browser in headless/hidden (default), 'virtual' screen mode, or headful/visible mode.
:param block_images: Prevent the loading of images through Firefox preferences.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param disable_resources: Drop requests of unnecessary resources for a speed boost. It depends but it made requests ~25% faster in my tests for some websites.
Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param block_webrtc: Blocks WebRTC entirely.
:param addons: List of Firefox addons to use. Must be paths to extracted addons.
:param humanize: Humanize the cursor movement. Takes either True or the MAX duration in seconds of the cursor movement. The cursor typically takes up to 1.5 seconds to move across the window.
:param allow_webgl: Whether to allow WebGL. To prevent leaks, only use this for special cases.
:param network_idle: Wait for the page until there are no network connections for at least 500 ms.
:param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000
:param page_action: Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again.
:param wait_selector: Wait for a specific css selector to be in a specific state.
:param wait_selector_state: The state to wait for the selector given with `wait_selector`. Default state is `attached`.
:param google_search: Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name.
:param extra_headers: A dictionary of extra headers to add to the request. _The referer set by the `google_search` argument takes priority over the referer set here if used together._
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
engine = CamoufoxEngine(
timeout=timeout,
headless=headless,
page_action=page_action,
block_images=block_images,
block_webrtc=block_webrtc,
addons=addons,
humanize=humanize,
allow_webgl=allow_webgl,
disable_resources=disable_resources,
network_idle=network_idle,
wait_selector=wait_selector,
wait_selector_state=wait_selector_state,
google_search=google_search,
extra_headers=extra_headers,
adaptor_arguments=self.adaptor_arguments,
)
return engine.fetch(url)
class PlayWrightFetcher(BaseFetcher):
"""A `Fetcher` class type that provide many options, all of them are based on PlayWright.
Using this Fetcher class, you can do requests with:
- Vanilla Playwright without any modifications other than the ones you chose.
- Stealthy Playwright with the stealth mode I wrote for it. It's still a work in progress but it bypasses many online tests like bot.sannysoft.com
Some of the things stealth mode does include:
1) Patches the CDP runtime fingerprint.
2) Mimics some of the real browsers' properties by injecting several JS files and using custom options.
3) Using custom flags on launch to hide Playwright even more and make it faster.
4) Generates real browser's headers of the same type and same user OS then append it to the request.
- Real browsers by passing the CDP URL of your browser to be controlled by the Fetcher and most of the options can be enabled on it.
- NSTBrowser's docker browserless option by passing the CDP URL and enabling `nstbrowser_mode` option.
> Note that these are the main options with PlayWright but it can be mixed together.
"""
def fetch(
self, url: str, headless: Union[bool, str] = True, disable_resources: bool = None,
useragent: Optional[str] = None, network_idle: Optional[bool] = False, timeout: Optional[float] = 30000,
page_action: Callable = do_nothing, wait_selector: Optional[str] = None, wait_selector_state: Optional[str] = 'attached',
hide_canvas: bool = True, disable_webgl: bool = False, extra_headers: Optional[Dict[str, str]] = None, google_search: Optional[bool] = True,
stealth: bool = False,
cdp_url: Optional[str] = None,
nstbrowser_mode: bool = False, nstbrowser_config: Optional[Dict] = None,
) -> Response:
"""Opens up a browser and do your request based on your chosen options below.
:param url: Target url.
:param headless: Run the browser in headless/hidden (default), or headful/visible mode.
:param disable_resources: Drop requests of unnecessary resources for speed boost. It depends but it made requests ~25% faster in my tests for some websites.
Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
This can help save your proxy usage but be careful with this option as it makes some websites never finish loading.
:param useragent: Pass a useragent string to be used. Otherwise the fetcher will generate a real Useragent of the same browser and use it.
:param network_idle: Wait for the page until there are no network connections for at least 500 ms.
:param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30000
:param page_action: Added for automation. A function that takes the `page` object, does the automation you need, then returns `page` again.
:param wait_selector: Wait for a specific css selector to be in a specific state.
:param wait_selector_state: The state to wait for the selector given with `wait_selector`. Default state is `attached`.
:param stealth: Enables stealth mode, check the documentation to see what stealth mode does currently.
:param hide_canvas: Add random noise to canvas operations to prevent fingerprinting.
:param disable_webgl: Disables WebGL and WebGL 2.0 support entirely.
:param google_search: Enabled by default, Scrapling will set the referer header to be as if this request came from a Google search for this website's domain name.
:param extra_headers: A dictionary of extra headers to add to the request. _The referer set by the `google_search` argument takes priority over the referer set here if used together._
:param cdp_url: Instead of launching a new browser instance, connect to this CDP URL to control real browsers/NSTBrowser through CDP.
:param nstbrowser_mode: Enables NSTBrowser mode, it have to be used with `cdp_url` argument or it will get completely ignored.
:param nstbrowser_config: The config you want to send with requests to the NSTBrowser. If left empty, Scrapling defaults to an optimized NSTBrowser's docker browserless config.
:return: A Response object with `url`, `text`, `content`, `status`, `reason`, `encoding`, `cookies`, `headers`, `request_headers`, and the `adaptor` class for parsing, of course.
"""
engine = PlaywrightEngine(
timeout=timeout,
stealth=stealth,
cdp_url=cdp_url,
headless=headless,
useragent=useragent,
page_action=page_action,
hide_canvas=hide_canvas,
network_idle=network_idle,
google_search=google_search,
extra_headers=extra_headers,
wait_selector=wait_selector,
disable_webgl=disable_webgl,
nstbrowser_mode=nstbrowser_mode,
nstbrowser_config=nstbrowser_config,
disable_resources=disable_resources,
wait_selector_state=wait_selector_state,
adaptor_arguments=self.adaptor_arguments,
)
return engine.fetch(url)
class CustomFetcher(BaseFetcher):
def fetch(self, url: str, browser_engine, **kwargs) -> Response:
engine = check_if_engine_usable(browser_engine)(adaptor_arguments=self.adaptor_arguments, **kwargs)
return engine.fetch(url)
+216 -51
View File
@@ -1,18 +1,14 @@
import os
import re
import inspect
from difflib import SequenceMatcher
from typing import Any, Dict, List, Tuple, Optional, Pattern, Union, Callable, Generator
try:
from typing import SupportsIndex
except ImportError:
# 'SupportsIndex' got added in Python 3.8
SupportsIndex = None
from scrapling.translator import HTMLTranslator
from scrapling.mixins import SelectorsGeneration
from scrapling.custom_types import TextHandler, AttributesHandler
from scrapling.storage_adaptors import SQLiteStorageSystem, StorageSystemMixin, _StorageTools
from scrapling.utils import setup_basic_logging, logging, clean_spaces, flatten, html_forbidden
from scrapling.core.translator import HTMLTranslator
from scrapling.core.mixins import SelectorsGeneration
from scrapling.core.custom_types import TextHandler, TextHandlers, AttributesHandler
from scrapling.core.storage_adaptors import SQLiteStorageSystem, StorageSystemMixin, _StorageTools
from scrapling.core.utils import setup_basic_logging, logging, clean_spaces, flatten, html_forbidden
from scrapling.core._types import Any, Dict, List, Tuple, Optional, Pattern, Union, Callable, Generator, SupportsIndex, Iterable
from lxml import etree, html
from cssselect import SelectorError, SelectorSyntaxError, parse as split_selectors
@@ -32,7 +28,7 @@ class Adaptor(SelectorsGeneration):
huge_tree: bool = True,
root: Optional[html.HtmlElement] = None,
keep_comments: Optional[bool] = False,
auto_match: Optional[bool] = False,
auto_match: Optional[bool] = True,
storage: Any = SQLiteStorageSystem,
storage_args: Optional[Dict] = None,
debug: Optional[bool] = True,
@@ -125,7 +121,7 @@ class Adaptor(SelectorsGeneration):
def _is_text_node(element: Union[html.HtmlElement, etree._ElementUnicodeResult]) -> bool:
"""Return True if given element is a result of a string expression
Examples:
Xpath -> '/text()', '/@attribute' etc...
XPath -> '/text()', '/@attribute' etc...
CSS3 -> '::text', '::attr(attrib)'...
"""
# Faster than checking `element.is_attribute or element.is_text or element.is_tail`
@@ -163,6 +159,8 @@ class Adaptor(SelectorsGeneration):
results = [self.__get_correct_result(n) for n in result]
if all(isinstance(res, self.__class__) for res in results):
return Adaptors(results)
elif all(isinstance(res, TextHandler) for res in results):
return TextHandlers(results)
return results
return self.__get_correct_result(result)
@@ -399,6 +397,56 @@ class Adaptor(SelectorsGeneration):
return self.__convert_results(score_table[highest_probability])
return []
def css_first(self, selector: str, identifier: str = '',
auto_match: bool = False, auto_save: bool = False, percentage: int = 0
) -> Union['Adaptor', 'TextHandler', None]:
"""Search current tree with CSS3 selectors and return the first result if possible, otherwise return `None`
**Important:
It's recommended to use the identifier argument if you plan to use different selector later
and want to relocate the same element(s)**
:param selector: The CSS3 selector to be used.
:param auto_match: Enabled will make function try to relocate the element if it was 'saved' before
:param identifier: A string that will be used to save/retrieve element's data in auto-matching
otherwise the selector will be used.
:param auto_save: Automatically save new elements for `auto_match` later
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
Be aware that the percentage calculation depends solely on the page structure so don't play with this
number unless you must know what you are doing!
:return: List as :class:`Adaptors`
"""
for element in self.css(selector, identifier, auto_match, auto_save, percentage):
return element
return None
def xpath_first(self, selector: str, identifier: str = '',
auto_match: bool = False, auto_save: bool = False, percentage: int = 0, **kwargs: Any
) -> Union['Adaptor', 'TextHandler', None]:
"""Search current tree with XPath selectors and return the first result if possible, otherwise return `None`
**Important:
It's recommended to use the identifier argument if you plan to use different selector later
and want to relocate the same element(s)**
Note: **Additional keyword arguments will be passed as XPath variables in the XPath expression!**
:param selector: The XPath selector to be used.
:param auto_match: Enabled will make function try to relocate the element if it was 'saved' before
:param identifier: A string that will be used to save/retrieve element's data in auto-matching
otherwise the selector will be used.
:param auto_save: Automatically save new elements for `auto_match` later
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
Be aware that the percentage calculation depends solely on the page structure so don't play with this
number unless you must know what you are doing!
:return: List as :class:`Adaptors`
"""
for element in self.xpath(selector, identifier, auto_match, auto_save, percentage, **kwargs):
return element
return None
def css(self, selector: str, identifier: str = '',
auto_match: bool = False, auto_save: bool = False, percentage: int = 0
) -> Union['Adaptors[Adaptor]', List]:
@@ -495,6 +543,113 @@ class Adaptor(SelectorsGeneration):
except (SelectorError, SelectorSyntaxError, etree.XPathError, etree.XPathEvalError):
raise SelectorSyntaxError(f"Invalid XPath selector: {selector}")
def find_all(self, *args: Union[str, Iterable[str], Pattern, Callable, Dict[str, str]], **kwargs: str) -> Union['Adaptors[Adaptor]', List]:
"""Find elements by filters of your creations for ease..
:param args: Tag name(s), an iterable of tag names, regex patterns, function, or a dictionary of elements' attributes. Leave empty for selecting all.
:param kwargs: The attributes you want to filter elements based on it.
:return: The `Adaptors` object of the elements or empty list
"""
# Attributes that are Python reserved words and can't be used directly
# Ex: find_all('a', class="blah") -> find_all('a', class_="blah")
# https://www.w3schools.com/python/python_ref_keywords.asp
whitelisted = {
'class_': 'class',
'for_': 'for',
}
if not args and not kwargs:
raise TypeError('You have to pass something to search with, like tag name(s), tag attributes, or both.')
attributes = dict()
tags, patterns = set(), set()
results, functions, selectors = [], [], []
def _search_tree(element: Adaptor, filter_function: Callable) -> None:
"""Collect element if it fulfills passed function otherwise, traverse the children tree and iterate"""
if filter_function(element):
results.append(element)
for branch in element.children:
_search_tree(branch, filter_function)
# Brace yourself for a wonderful journey!
for arg in args:
if type(arg) is str:
tags.add(arg)
elif type(arg) in [list, tuple, set]:
if not all(map(lambda x: type(x) is str, arg)):
raise TypeError('Nested Iterables are not accepted, only iterables of tag names are accepted')
tags.update(set(arg))
elif type(arg) is dict:
if not all([(type(k) is str and type(v) is str) for k, v in arg.items()]):
raise TypeError('Nested dictionaries are not accepted, only string keys and string values are accepted')
attributes.update(arg)
elif type(arg) is re.Pattern:
patterns.add(arg)
elif callable(arg):
if len(inspect.signature(arg).parameters) > 0:
functions.append(arg)
else:
raise TypeError("Callable filter function must have at least one argument to take `Adaptor` objects.")
else:
raise TypeError(f'Argument with type "{type(arg)}" is not accepted, please read the docs.')
if not all([(type(k) is str and type(v) is str) for k, v in kwargs.items()]):
raise TypeError('Only string values are accepted for arguments')
for attribute_name, value in kwargs.items():
# Only replace names for kwargs, replacing them in dictionaries doesn't make sense
attribute_name = whitelisted.get(attribute_name, attribute_name)
attributes[attribute_name] = value
# It's easier and faster to build a selector than traversing the tree
tags = tags or ['']
for tag in tags:
selector = tag
for key, value in attributes.items():
value = value.replace('"', r'\"') # Escape double quotes in user input
# Not escaping anything with the key so the user can pass patterns like {'href*': '/p/'} or get errors :)
selector += '[{}="{}"]'.format(key, value)
if selector:
selectors.append(selector)
if selectors:
results = self.css(', '.join(selectors))
if results:
# From the results, get the ones that fulfill passed regex patterns
for pattern in patterns:
results = results.filter(lambda e: e.text.re(pattern, check_match=True))
# From the results, get the ones that fulfill passed functions
for function in functions:
results = results.filter(function)
else:
for pattern in patterns:
results.extend(self.find_by_regex(pattern, first_match=False))
for result in (results or [self]):
for function in functions:
_search_tree(result, function)
return self.__convert_results(results)
def find(self, *args: Union[str, Iterable[str], Pattern, Callable, Dict[str, str]], **kwargs: str) -> Union['Adaptor', None]:
"""Find elements by filters of your creations for ease then return the first result. Otherwise return `None`.
:param args: Tag name(s), an iterable of tag names, regex patterns, function, or a dictionary of elements' attributes. Leave empty for selecting all.
:param kwargs: The attributes you want to filter elements based on it.
:return: The `Adaptor` object of the element or `None` if the result didn't match
"""
for element in self.find_all(*args, **kwargs):
return element
return None
def __calculate_similarity_score(self, original: Dict, candidate: html.HtmlElement) -> float:
"""Used internally to calculate a score that shows how candidate element similar to the original one
@@ -606,25 +761,33 @@ class Adaptor(SelectorsGeneration):
# Operations on text functions
def json(self) -> Dict:
"""Return json response if the response is jsonable otherwise throws error"""
return self.text.json()
if self.text:
return self.text.json()
else:
return self.get_all_text(strip=True).json()
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True) -> 'List[str]':
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False) -> 'List[str]':
"""Apply the given regex to the current text and return a list of strings with the matches.
:param regex: Can be either a compiled regular expression or a string.
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
return self.text.re(regex, replace_entities)
return self.text.re(regex, replace_entities, clean_match, case_sensitive)
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True):
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False) -> Union[str, None]:
"""Apply the given regex to text and return the first match if found, otherwise return the default value.
:param regex: Can be either a compiled regular expression or a string.
:param default: The default value to be returned if there is no match
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
return self.text.re_first(regex, default, replace_entities)
return self.text.re_first(regex, default, replace_entities, clean_match, case_sensitive)
def find_similar(
self,
@@ -757,10 +920,10 @@ class Adaptor(SelectorsGeneration):
return self.__convert_results(results)
def find_by_regex(
self, query: str, first_match: bool = True, case_sensitive: bool = False, clean_match: bool = True
self, query: Union[str, Pattern[str]], first_match: bool = True, case_sensitive: bool = False, clean_match: bool = True
) -> Union['Adaptors[Adaptor]', 'Adaptor', List]:
"""Find elements that its text content matches the input regex pattern.
:param query: Regex query to match
:param query: Regex query/pattern to match
:param first_match: Return first element that matches conditions, enabled by default
:param case_sensitive: if enabled, letters case will be taken into consideration in the regex
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
@@ -855,54 +1018,56 @@ class Adaptors(List[Adaptor]):
]
return self.__class__(flatten(results))
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True) -> 'List[str]':
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False) -> 'List[str]':
"""Call the ``.re()`` method for each element in this list and return
their results flattened as List of TextHandler.
:param regex: Can be either a compiled regular expression or a string.
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
results = [
n.text.re(regex, replace_entities) for n in self
n.text.re(regex, replace_entities, clean_match, case_sensitive) for n in self
]
return flatten(results)
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True):
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True,
clean_match: bool = False, case_sensitive: bool = False) -> Union[str, None]:
"""Call the ``.re_first()`` method for each element in this list and return
their results flattened as List of TextHandler.
the first result or the default value otherwise.
:param regex: Can be either a compiled regular expression or a string.
:param default: The default value to be returned if there is no match
:param replace_entities: if enabled character entity references are replaced by their corresponding character
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
"""
for n in self:
for result in n.re(regex, replace_entities, clean_match, case_sensitive):
return result
return default
def search(self, func: Callable[['Adaptor'], bool]) -> Union['Adaptor', None]:
"""Loop over all current elements and return the first element that matches the passed function
:param func: A function that takes each element as an argument and returns True/False
:return: The first element that match the function or ``None`` otherwise.
"""
for element in self:
if func(element):
return element
return None
def filter(self, func: Callable[['Adaptor'], bool]) -> Union['Adaptors', List]:
"""Filter current elements based on the passed function
:param func: A function that takes each element as an argument and returns True/False
:return: The new `Adaptors` object or empty list otherwise.
"""
results = [
n.text.re_first(regex, default, replace_entities) for n in self
element for element in self if func(element)
]
return flatten(results)
# def __getattr__(self, name):
# if name in dir(self.__class__):
# return super().__getattribute__(name)
#
# # Execute the method itself on each Adaptor
# results = []
# for item in self:
# results.append(getattr(item, name))
#
# if all(callable(r) for r in results):
# def call_all(*args, **kwargs):
# final_results = [r(*args, **kwargs) for r in results]
# if all([isinstance(r, (Adaptor, Adaptors,)) for r in results]):
# return self.__class__(final_results)
# return final_results
#
# return call_all
# else:
# # Flatten the result if it's a single-item list containing a list
# if len(self) == 1 and isinstance(results[0], list):
# return self.__class__(results[0])
# return self.__class__(results)
return self.__class__(results) if results else results
def get(self, default=None):
"""Returns the first item of the current list
+2 -2
View File
@@ -1,8 +1,8 @@
[metadata]
name = scrapling
version = 0.1.2
version = 0.2
author = Karim Shoair
author_email = karim.shoair@pm.me
description = Scrapling is a powerful, flexible, adaptive, and high-performance web scraping library for Python.
license = BSD
home-page = https://github.com/D4Vinci/Scrapling
home_page = https://github.com/D4Vinci/Scrapling
+11 -5
View File
@@ -1,4 +1,4 @@
from setuptools import setup
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
@@ -6,7 +6,7 @@ with open("README.md", "r", encoding="utf-8") as fh:
setup(
name="scrapling",
version="0.1.2",
version="0.2",
description="""Scrapling is a powerful, flexible, and high-performance web scraping library for Python. It
simplifies the process of extracting data from websites, even when they undergo structural changes, and offers
impressive speed improvements over many popular scraping tools.""",
@@ -15,7 +15,7 @@ setup(
author="Karim Shoair",
author_email="karim.shoair@pm.me",
license="BSD",
packages=["scrapling",],
packages=find_packages(),
zip_safe=False,
package_dir={
"scrapling": "scrapling",
@@ -32,16 +32,17 @@ setup(
"Natural Language :: English",
"Topic :: Internet :: WWW/HTTP",
"Topic :: Text Processing :: Markup",
"Topic :: Internet :: WWW/HTTP :: Browsers",
"Topic :: Text Processing :: Markup :: HTML",
"Topic :: Software Development :: Libraries :: Python Modules",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: Implementation :: CPython",
"Typing :: Typed",
],
@@ -53,8 +54,13 @@ setup(
"w3lib",
"orjson>=3",
"tldextract",
'httpx[brotli,zstd]',
'playwright',
'rebrowser-playwright',
'camoufox>=0.3.7',
'browserforge',
],
python_requires=">=3.7",
python_requires=">=3.8",
url="https://github.com/D4Vinci/Scrapling",
project_urls={
"Documentation": "https://github.com/D4Vinci/Scrapling/tree/main/docs", # For now
+1
View File
@@ -0,0 +1 @@
# Because I'm too lazy to mock requests :)
+62
View File
@@ -0,0 +1,62 @@
import unittest
import pytest_httpbin
from scrapling import StealthyFetcher
@pytest_httpbin.use_class_based_httpbin
# @pytest_httpbin.use_class_based_httpbin_secure
class TestStealthyFetcher(unittest.TestCase):
def setUp(self):
self.fetcher = StealthyFetcher(auto_match=False)
url = self.httpbin.url
self.status_200 = f'{url}/status/200'
self.status_404 = f'{url}/status/404'
self.status_501 = f'{url}/status/501'
self.basic_url = f'{url}/get'
self.html_url = f'{url}/html'
self.delayed_url = f'{url}/delay/10' # 10 Seconds delay response
self.cookies_url = f"{url}/cookies/set/test/value"
def test_basic_fetch(self):
"""Test doing basic fetch request with multiple statuses"""
self.assertEqual(self.fetcher.fetch(self.status_200).status, 200)
self.assertEqual(self.fetcher.fetch(self.status_404).status, 404)
self.assertEqual(self.fetcher.fetch(self.status_501).status, 501)
def test_networkidle(self):
"""Test if waiting for `networkidle` make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.basic_url, network_idle=True).status, 200)
def test_blocking_resources(self):
"""Test if blocking resources make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.basic_url, block_images=True).status, 200)
self.assertEqual(self.fetcher.fetch(self.basic_url, disable_resources=True).status, 200)
def test_waiting_selector(self):
"""Test if waiting for a selector make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.html_url, wait_selector='h1').status, 200)
def test_cookies_loading(self):
"""Test if cookies are set after the request"""
self.assertEqual(self.fetcher.fetch(self.cookies_url).cookies, {'test': 'value'})
def test_automation(self):
"""Test if automation break the code or not"""
def scroll_page(page):
page.mouse.wheel(10, 0)
page.mouse.move(100, 400)
page.mouse.up()
return page
self.assertEqual(self.fetcher.fetch(self.html_url, page_action=scroll_page).status, 200)
def test_properties(self):
"""Test if different arguments breaks the code or not"""
self.assertEqual(self.fetcher.fetch(self.html_url, block_webrtc=True, allow_webgl=True).status, 200)
self.assertEqual(self.fetcher.fetch(self.html_url, block_webrtc=False, allow_webgl=True).status, 200)
self.assertEqual(self.fetcher.fetch(self.html_url, block_webrtc=True, allow_webgl=False).status, 200)
def test_infinite_timeout(self):
"""Test if infinite timeout breaks the code or not"""
self.assertEqual(self.fetcher.fetch(self.delayed_url, timeout=None).status, 200)
+67
View File
@@ -0,0 +1,67 @@
import unittest
import pytest_httpbin
from scrapling import Fetcher
@pytest_httpbin.use_class_based_httpbin
class TestFetcher(unittest.TestCase):
def setUp(self):
self.fetcher = Fetcher(auto_match=False)
url = self.httpbin.url
self.status_200 = f'{url}/status/200'
self.status_404 = f'{url}/status/404'
self.status_501 = f'{url}/status/501'
self.basic_url = f'{url}/get'
self.post_url = f'{url}/post'
self.put_url = f'{url}/put'
self.delete_url = f'{url}/delete'
self.html_url = f'{url}/html'
def test_basic_get(self):
"""Test doing basic get request with multiple statuses"""
self.assertEqual(self.fetcher.get(self.status_200).status, 200)
self.assertEqual(self.fetcher.get(self.status_404).status, 404)
self.assertEqual(self.fetcher.get(self.status_501).status, 501)
def test_get_properties(self):
"""Test if different arguments with GET request breaks the code or not"""
self.assertEqual(self.fetcher.get(self.status_200, stealthy_headers=True).status, 200)
self.assertEqual(self.fetcher.get(self.status_200, follow_redirects=True).status, 200)
self.assertEqual(self.fetcher.get(self.status_200, timeout=None).status, 200)
self.assertEqual(
self.fetcher.get(self.status_200, stealthy_headers=True, follow_redirects=True, timeout=None).status,
200
)
def test_post_properties(self):
"""Test if different arguments with POST request breaks the code or not"""
self.assertEqual(self.fetcher.post(self.post_url, data={'key': 'value'}).status, 200)
self.assertEqual(self.fetcher.post(self.post_url, data={'key': 'value'}, stealthy_headers=True).status, 200)
self.assertEqual(self.fetcher.post(self.post_url, data={'key': 'value'}, follow_redirects=True).status, 200)
self.assertEqual(self.fetcher.post(self.post_url, data={'key': 'value'}, timeout=None).status, 200)
self.assertEqual(
self.fetcher.post(self.post_url, data={'key': 'value'}, stealthy_headers=True, follow_redirects=True, timeout=None).status,
200
)
def test_put_properties(self):
"""Test if different arguments with PUT request breaks the code or not"""
self.assertEqual(self.fetcher.put(self.put_url, data={'key': 'value'}).status, 200)
self.assertEqual(self.fetcher.put(self.put_url, data={'key': 'value'}, stealthy_headers=True).status, 200)
self.assertEqual(self.fetcher.put(self.put_url, data={'key': 'value'}, follow_redirects=True).status, 200)
self.assertEqual(self.fetcher.put(self.put_url, data={'key': 'value'}, timeout=None).status, 200)
self.assertEqual(
self.fetcher.put(self.put_url, data={'key': 'value'}, stealthy_headers=True, follow_redirects=True, timeout=None).status,
200
)
def test_delete_properties(self):
"""Test if different arguments with DELETE request breaks the code or not"""
self.assertEqual(self.fetcher.delete(self.delete_url, stealthy_headers=True).status, 200)
self.assertEqual(self.fetcher.delete(self.delete_url, follow_redirects=True).status, 200)
self.assertEqual(self.fetcher.delete(self.delete_url, timeout=None).status, 200)
self.assertEqual(
self.fetcher.delete(self.delete_url, stealthy_headers=True, follow_redirects=True, timeout=None).status,
200
)
+74
View File
@@ -0,0 +1,74 @@
import unittest
import pytest_httpbin
from scrapling import PlayWrightFetcher
@pytest_httpbin.use_class_based_httpbin
# @pytest_httpbin.use_class_based_httpbin_secure
class TestPlayWrightFetcher(unittest.TestCase):
def setUp(self):
self.fetcher = PlayWrightFetcher(auto_match=False)
url = self.httpbin.url
self.status_200 = f'{url}/status/200'
self.status_404 = f'{url}/status/404'
self.status_501 = f'{url}/status/501'
self.basic_url = f'{url}/get'
self.html_url = f'{url}/html'
self.delayed_url = f'{url}/delay/10' # 10 Seconds delay response
self.cookies_url = f"{url}/cookies/set/test/value"
def test_basic_fetch(self):
"""Test doing basic fetch request with multiple statuses"""
self.assertEqual(self.fetcher.fetch(self.status_200).status, 200)
self.assertEqual(self.fetcher.fetch(self.status_404).status, 404)
self.assertEqual(self.fetcher.fetch(self.status_501).status, 501)
def test_networkidle(self):
"""Test if waiting for `networkidle` make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.basic_url, network_idle=True).status, 200)
def test_blocking_resources(self):
"""Test if blocking resources make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.basic_url, disable_resources=True).status, 200)
def test_waiting_selector(self):
"""Test if waiting for a selector make page does not finish loading or not"""
self.assertEqual(self.fetcher.fetch(self.html_url, wait_selector='h1').status, 200)
def test_cookies_loading(self):
"""Test if cookies are set after the request"""
self.assertEqual(self.fetcher.fetch(self.cookies_url).cookies, {'test': 'value'})
def test_automation(self):
"""Test if automation break the code or not"""
def scroll_page(page):
page.mouse.wheel(10, 0)
page.mouse.move(100, 400)
page.mouse.up()
return page
self.assertEqual(self.fetcher.fetch(self.html_url, page_action=scroll_page).status, 200)
def test_properties(self):
"""Test if different arguments breaks the code or not"""
self.assertEqual(self.fetcher.fetch(self.html_url, disable_webgl=True, hide_canvas=False).status, 200)
self.assertEqual(self.fetcher.fetch(self.html_url, disable_webgl=False, hide_canvas=True).status, 200)
self.assertEqual(self.fetcher.fetch(self.html_url, stealth=True).status, 200)
self.assertEqual(self.fetcher.fetch(self.html_url, useragent='Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0').status, 200)
def test_cdp_url(self):
"""Test if it's going to try to connect to cdp url or not"""
with self.assertRaises(ValueError):
_ = self.fetcher.fetch(self.html_url, cdp_url='blahblah')
with self.assertRaises(ValueError):
_ = self.fetcher.fetch(self.html_url, cdp_url='blahblah', nstbrowser_mode=True)
with self.assertRaises(Exception):
# There's no type for this error in PlayWright, it's just `Error`
_ = self.fetcher.fetch(self.html_url, cdp_url='ws://blahblah')
def test_infinite_timeout(self):
"""Test if infinite timeout breaks the code or not"""
self.assertEqual(self.fetcher.fetch(self.delayed_url, timeout=None).status, 200)
View File
+56
View File
@@ -0,0 +1,56 @@
import unittest
from scrapling import Adaptor
class TestParserAutoMatch(unittest.TestCase):
def test_element_relocation(self):
"""Test relocating element after structure change"""
original_html = '''
<div class="container">
<section class="products">
<article class="product" id="p1">
<h3>Product 1</h3>
<p class="description">Description 1</p>
</article>
<article class="product" id="p2">
<h3>Product 2</h3>
<p class="description">Description 2</p>
</article>
</section>
</div>
'''
changed_html = '''
<div class="new-container">
<div class="product-wrapper">
<section class="products">
<article class="product new-class" data-id="p1">
<div class="product-info">
<h3>Product 1</h3>
<p class="new-description">Description 1</p>
</div>
</article>
<article class="product new-class" data-id="p2">
<div class="product-info">
<h3>Product 2</h3>
<p class="new-description">Description 2</p>
</div>
</article>
</section>
</div>
</div>
'''
old_page = Adaptor(original_html, url='example.com', auto_match=True, debug=True)
new_page = Adaptor(changed_html, url='example.com', auto_match=True, debug=True)
# 'p1' was used as ID and now it's not and all the path elements have changes
# Also at the same time testing auto-match vs combined selectors
_ = old_page.css('#p1, #p2', auto_save=True)[0]
relocated = new_page.css('#p1', auto_match=True)
self.assertIsNotNone(relocated)
self.assertEqual(relocated[0].attrib['data-id'], 'p1')
self.assertTrue(relocated[0].has_class('new-class'))
self.assertEqual(relocated[0].css('.new-description')[0].text, 'Description 1')
@@ -112,11 +112,11 @@ class TestParser(unittest.TestCase):
def test_find_similar_elements(self):
"""Test Finding similar elements of an element"""
first_product = self.page.css('.product')[0]
first_product = self.page.css_first('.product')
similar_products = first_product.find_similar()
self.assertEqual(len(similar_products), 2)
first_review = self.page.css('.review')[0]
first_review = self.page.find('div', class_='review')
similar_high_rated_reviews = [
review
for review in first_review.find_similar()
@@ -127,16 +127,16 @@ class TestParser(unittest.TestCase):
def test_expected_errors(self):
"""Test errors that should raised if it does"""
with self.assertRaises(ValueError):
_ = Adaptor()
_ = Adaptor(auto_match=False)
with self.assertRaises(TypeError):
_ = Adaptor(root="ayo")
_ = Adaptor(root="ayo", auto_match=False)
with self.assertRaises(TypeError):
_ = Adaptor(text=1)
_ = Adaptor(text=1, auto_match=False)
with self.assertRaises(TypeError):
_ = Adaptor(body=1)
_ = Adaptor(body=1, auto_match=False)
with self.assertRaises(ValueError):
_ = Adaptor(self.html, storage=object, auto_match=True)
@@ -169,8 +169,8 @@ class TestParser(unittest.TestCase):
def test_selectors_generation(self):
"""Try to create selectors for all elements in the page"""
def _traverse(element: Adaptor):
self.assertTrue(type(element.css_selector) is str)
self.assertTrue(type(element.xpath_selector) is str)
self.assertTrue(type(element.generate_css_selector) is str)
self.assertTrue(type(element.generate_xpath_selector) is str)
for branch in element.children:
_traverse(branch)
@@ -197,7 +197,7 @@ class TestParser(unittest.TestCase):
parent_siblings = parent.siblings
self.assertEqual(len(parent_siblings), 1)
child = table.css('[data-id="1"]')[0]
child = table.find({'data-id': "1"})
next_element = child.next
self.assertEqual(next_element.attrib['data-id'], '2')
@@ -261,60 +261,10 @@ class TestParser(unittest.TestCase):
key_value = list(products[0].attrib.search_values('1', partial=True))
self.assertEqual(list(key_value[0].keys()), ['data-id'])
attr_json = self.page.css('#products')[0].attrib['schema'].json()
attr_json = self.page.css_first('#products').attrib['schema'].json()
self.assertEqual(attr_json, {'jsonable': 'data'})
self.assertEqual(type(self.page.css('#products')[0].attrib.json_string), bytes)
def test_element_relocation(self):
"""Test relocating element after structure change"""
original_html = '''
<div class="container">
<section class="products">
<article class="product" id="p1">
<h3>Product 1</h3>
<p class="description">Description 1</p>
</article>
<article class="product" id="p2">
<h3>Product 2</h3>
<p class="description">Description 2</p>
</article>
</section>
</div>
'''
changed_html = '''
<div class="new-container">
<div class="product-wrapper">
<section class="products">
<article class="product new-class" data-id="p1">
<div class="product-info">
<h3>Product 1</h3>
<p class="new-description">Description 1</p>
</div>
</article>
<article class="product new-class" data-id="p2">
<div class="product-info">
<h3>Product 2</h3>
<p class="new-description">Description 2</p>
</div>
</article>
</section>
</div>
</div>
'''
old_page = Adaptor(original_html, url='example.com', auto_match=True, debug=True)
new_page = Adaptor(changed_html, url='example.com', auto_match=True, debug=True)
# 'p1' was used as ID and now it's not and all the path elements have changes
# Also at the same time testing auto-match vs combined selectors
_ = old_page.css('#p1, #p2', auto_save=True)[0]
relocated = new_page.css('#p1', auto_match=True)
self.assertIsNotNone(relocated)
self.assertEqual(relocated[0].attrib['data-id'], 'p1')
self.assertTrue(relocated[0].has_class('new-class'))
self.assertEqual(relocated[0].css('.new-description')[0].text, 'Description 1')
def test_performance(self):
"""Test parsing and selecting speed"""
import time
@@ -331,6 +281,6 @@ class TestParser(unittest.TestCase):
self.assertLess(end_time - start_time, 0.1)
# Use `coverage run -m unittest --verbose tests/test_all_functions.py` instead for the coverage report
# Use `coverage run -m unittest --verbose tests/test_parser_functions.py` instead for the coverage report
# if __name__ == '__main__':
# unittest.main(verbosity=2)
+7 -2
View File
@@ -1,2 +1,7 @@
pytest
pytest-cov
pytest>=2.8.0,<9
pytest-cov
playwright
camoufox
werkzeug<3.0.0
pytest-httpbin==2.1.0
httpbin~=0.10.0
+5 -2
View File
@@ -4,14 +4,17 @@
# and then run "tox" from this directory.
[tox]
envlist = pre-commit,py37,py38,py39,py310,py311,py312
envlist = pre-commit,py{38,39,310,311,312,313}
[testenv]
usedevelop = True
changedir = tests
deps =
-r{toxinidir}/tests/requirements.txt
commands = pytest --cov=scrapling --cov-report=xml
commands =
playwright install-deps chromium firefox
camoufox fetch --browserforge
pytest --cov=scrapling --cov-report=xml
[testenv:pre-commit]
basepython = python3