From 201e0665c35244df3fe2e673073233e949bdfcb9 Mon Sep 17 00:00:00 2001
From: johanbeytell1 <107624055+johanbeytell1@users.noreply.github.com>
Date: Sat, 31 Oct 2026 11:57:05 +0000
Subject: [PATCH 1/2] Release 0.1.8: security hardening, bug fixes, offline
tests and CI
Security
- Timeouts on every request (default (10, 30) seconds, overridable).
- Response size caps (max_page_bytes, max_image_bytes), enforced from
Content-Length and while streaming.
- Redirects followed hop by hop, each hop validated; max_redirects bounds
the chain. Only http/https are fetched.
- Child links must stay on the same domain after redirects.
- New block_private_addresses flag (default False) to refuse non-public
targets, applied to pages and images.
- CSV formula injection neutralised.
- Safe, unique image file names; corrupt and decompression-bomb images
are skipped instead of crashing the run.
- requests>=2.32.4 (CVE-2024-47081), pillow>=10.3.0 (CVE-2024-28219).
- CI: actions pinned to commit SHAs, least-privilege permissions, no
persisted checkout credentials, PyPI token only in the publish job.
Bug fixes
- Inline markup no longer glues words; nested containers no longer
repeat text; head/title/script/style/template/comments excluded;
yields line breaks.
- links mode returns img/video/audio/source URLs, absolute, http(s) only.
- unseen mode includes script data.
- 4xx/5xx pages are skipped instead of being scraped.
- Each page and each image is fetched once.
- Images without src are skipped; empty image names no longer crash.
- rate_limit is shared across threads.
- Fragments and host/port variants are de-duplicated.
- Output is deterministic (discovery order); de-duplication runs after the
crawl. Duplicate similarity measured against the larger text, so short
items no longer swallow unrelated paragraphs.
- Failed pages are skipped and logged (logger 'pywebscrapr') instead of
aborting the run and discarding output.
- Links files: UTF-8 (BOM tolerated), blank lines ignored.
- Charset taken from HTTP header or meta tag; honoured.
- max_links_to_follow counts child links only; 0 means no child links.
- Invalid options raise ValueError early.
Packaging, CI and tests
- Version defined once in pywebscrapr/__init__.py (__version__).
- beautifulsoup4 replaces bs4; Python >=3.9.
- Workflow tests on 3.9/3.12/3.13 for pushes and PRs; publishing runs
only for tags that pass the tests and match __version__.
- tests.py rewritten as an offline suite on local fixture servers, plus an
opt-in live test (PYWEBSCRAPR_LIVE_TESTS=1).
- README: consolidated 0.1.8 changelog and usage notes.
Co-authored-by: arena-agent <297053741+arena-agent@users.noreply.github.com>
---
.github/workflows/build_and_publish.yml | 84 +-
.gitignore | 4 +-
README.md | 111 ++-
pywebscrapr/__init__.py | 987 +++++++++++++++++-------
requirements.txt | 6 +-
setup.py | 28 +-
tests.py | 510 +++++++++++-
7 files changed, 1385 insertions(+), 345 deletions(-)
diff --git a/.github/workflows/build_and_publish.yml b/.github/workflows/build_and_publish.yml
index 4a9ec6b..71d7d52 100644
--- a/.github/workflows/build_and_publish.yml
+++ b/.github/workflows/build_and_publish.yml
@@ -1,34 +1,64 @@
-name: Publish Python 🐍 distributions 📦 to PyPI
+name: Test and publish to PyPI
on:
push:
- tags:
- - '*'
+ branches: [main]
+ tags: ['*']
+ pull_request:
+
+# Least privilege by default. Jobs that need more say so explicitly.
+permissions:
+ contents: read
jobs:
- build-n-publish:
- name: Build and publish Python 🐍 distributions 📦 to PyPI
+ test:
+ name: Test (Python ${{ matrix.python-version }})
+ runs-on: ubuntu-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ python-version: ['3.9', '3.12', '3.13']
+ steps:
+ - name: Check out code
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0
+ with:
+ python-version: ${{ matrix.python-version }}
+ - name: Install dependencies
+ run: python -m pip install -r requirements.txt
+ - name: Run tests (offline)
+ run: python -m unittest -v tests
+
+ build-and-publish:
+ name: Build and publish to PyPI
+ needs: test
+ # Only tag pushes publish. Pull requests and branch pushes stop after the tests.
+ if: startsWith(github.ref, 'refs/tags/')
runs-on: ubuntu-latest
steps:
- - uses: actions/checkout@master
- - name: Set up Python 3.12
- uses: actions/setup-python@v3
- with:
- python-version: '3.12'
- - name: Install pypa/setuptools
- run: >-
- python -m
- pip install setuptools wheel
- - name: Extract tag name
- id: tag
- run: echo ::set-output name=TAG_NAME::$(echo $GITHUB_REF | cut -d / -f 3)
- - name: Update version in setup.py
- run: >-
- sed -i "s/{{VERSION_PLACEHOLDER}}/${{ steps.tag.outputs.TAG_NAME }}/g" setup.py
- - name: Build a binary wheel
- run: >-
- python setup.py sdist bdist_wheel
- - name: Publish distribution 📦 to PyPI
- uses: pypa/gh-action-pypi-publish@master
- with:
- password: ${{ secrets.PYPI_API_TOKEN }}
\ No newline at end of file
+ - name: Check out code
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
+ - name: Set up Python
+ uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0
+ with:
+ python-version: '3.12'
+ - name: Install dependencies
+ run: python -m pip install -r requirements.txt
+ - name: Check that the release tag matches the package version
+ env:
+ TAG_NAME: ${{ github.ref_name }}
+ run: |
+ python -c "import os, pywebscrapr; tag = os.environ['TAG_NAME']; version = pywebscrapr.__version__; assert tag == version, 'tag %s does not match __version__ %s' % (tag, version); print('Releasing', version)"
+ - name: Build sdist and wheel
+ run: |
+ python -m pip install build
+ python -m build
+ - name: Publish distribution to PyPI
+ uses: pypa/gh-action-pypi-publish@7245906fb61fa33a13554f9d5fb856fe61758086 # v1.14.3
+ with:
+ password: ${{ secrets.PYPI_API_TOKEN }}
diff --git a/.gitignore b/.gitignore
index 7391daf..3a70787 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,4 +4,6 @@ dist/
build/
venv/
json_output.json
-output.txt
\ No newline at end of file
+output.txt
+output/
+*.egg-info/
diff --git a/README.md b/README.md
index 518cc35..ef1fab8 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,71 @@
An open-source Python library for web scraping tasks. Includes support for both text and image scraping.
+## Changes in 0.1.8
+
+0.1.8 is a security, reliability and correctness release. It changes some defaults and behaviour, so read **Breaking changes** before upgrading.
+
+### Breaking changes
+- **Python 3.9 or later is required** (the package previously claimed 3.6+). Pillow 10.3+ and requests 2.32.4+ are the new minimums, and the CI matrix tests 3.9, 3.12 and 3.13.
+- **Failed pages and images are skipped and logged instead of raising.** A bad URL, timeout, DNS failure, HTTP error, oversized response or corrupt image no longer aborts the run or discards the output. Failures are logged as warnings and downloads as info messages on the `pywebscrapr` logger. Previously "Downloaded" and "Ignored" messages were printed; to see them again, use `logging.basicConfig(level=logging.INFO)`.
+- **Text output changes.** Words are no longer glued together, nested containers no longer repeat their text, and ``, ``, `<style>`, `<template>` and comments are excluded. Expect different (cleaner) output from earlier versions.
+- **Duplicate detection measures similarity against the larger of the two paragraphs.** The old measure was against the earlier paragraph, which let a short item such as "Home" delete unrelated content. Results for paragraphs of similar length are essentially unchanged.
+- **`max_links_to_follow` counts child links only.** Seed links are always scraped. Previously seeds counted toward the limit and `0` meant "unlimited"; now `0` means "follow no child links".
+- **Child links must stay on the same domain after redirects.** Seed links may still redirect anywhere.
+- **Result URLs are final URLs.** The `URL` field in CSV and JSON output is the address after redirects.
+- **Default User-Agent.** Requests now send `PyWebScrapr/0.1.8 (+https://github.com/infinitode/pywebscrapr)`. Pass `user_agent=None` to keep the requests default.
+- **Invalid options raise `ValueError` straight away**, for example an unknown `elements_to_scrape`, `max_workers < 1`, or a non-positive timeout.
+- **Image file extensions match the detected format** (for example `photo.jpeg` is saved as `photo.jpg` when the image really is a JPEG).
+- **Links files are read as UTF-8** (a byte-order mark is tolerated). Blank lines and surrounding whitespace are ignored.
+- **`output_file` can be `None`** to skip the plain-text file. The default is still `output.txt`.
+- Private helpers `_download_image` and `_check_image_dimensions` were replaced. They were never documented, but they are no longer importable under those names.
+
+### Security
+- **Timeouts on every request.** The default is `timeout=(10, 30)` (connect, read seconds). Previously a server that never answered could hang the run forever. Pass `timeout=None` only if you accept that risk.
+- **Response size limits.** `max_page_bytes` (default 10 MB) and `max_image_bytes` (default 20 MB) are enforced from `Content-Length` and again while the body streams. Previously the whole response was loaded into memory with no cap.
+- **Redirects are checked hop by hop.** Each redirect target is validated for scheme and, when enabled, address. `max_redirects` (default 10) bounds the chain.
+- **Only `http` and `https` are fetched.** `file:`, `ftp:`, `javascript:` and other schemes are refused.
+- **New `block_private_addresses` flag** (default `False`, so private and internal hosts are allowed as before). Set it to `True` to refuse URLs that resolve to loopback, private, link-local, CGNAT or other non-public addresses. This protects against a crawled page pointing the scraper at internal services, such as an `<img src="http://169.254.169.254/...">`. Limitation: the check runs before the connection is made, so it does not defend against DNS rebinding; keep the flag on for untrusted inputs and run the scraper with network egress controls where that matters.
+- **CSV formula injection is neutralised.** Cells beginning with `=`, `+`, `-`, `@`, tab or carriage return get a leading apostrophe, so spreadsheet programs will not evaluate scraped text as formulas.
+- **Safe image file names.** Names are restricted to letters, digits, `.`, `_` and `-`; Windows reserved names are prefixed; empty names fall back to `image`. Two images that would share a name in the same run are stored under different names instead of overwriting each other. Files left over from an earlier run with the same name are still overwritten, as before.
+- **Dependency floors.** `requests>=2.32.4` includes the fix for CVE-2024-47081 (a `.netrc` credential leak). `pillow>=10.3.0` includes the fix for CVE-2024-28219 (a buffer overflow in `_imagingcms`).
+- **CI hardening.** Every GitHub Action is pinned to a commit SHA. Workflows default to `permissions: contents: read`, checkout does not persist credentials, and the PyPI token is only available to the publish job.
+
+### Bug fixes
+- **Text extraction** joins inline markup correctly: `Hello <b>world</b>` gives `Hello world`, not `Helloworld`.
+- **Nested containers no longer duplicate text.** `<div><div><p>x</p></div></div>` yields `x` once, not three times.
+- **Line breaks.** `<br>` produces a line break in text output.
+- **`links` mode** now returns images, video, audio and `<source>` URLs as well as anchors (it previously returned only anchors, and the image and video branches never ran). URLs are returned absolute, not relative, and only `http`/`https` URLs are kept.
+- **`unseen` mode** includes `<script>` data as documented. It previously read only `<meta>` tags.
+- **Error pages are skipped.** 4xx and 5xx responses are no longer saved as page content.
+- **Each page is fetched once.** With `follow_child_links`, a page was previously downloaded twice (once for its content and once to find its links).
+- **Each image is downloaded once.** Images were previously requested twice (once to check the size and once to save). The check and the save could also see different responses.
+- **Images without a `src` are skipped.** They previously resolved to the page's own URL and triggered extra requests.
+- **Image names no longer crash the run.** A URL with an empty file name (such as `/?format=png`) used to raise `IsADirectoryError` and abort everything.
+- **Decompression bombs and corrupt images are skipped.** A header declaring an enormous image previously raised `DecompressionBombError` outside the error handling and aborted the run.
+- **Rate limit is global.** `rate_limit` is now the minimum spacing between requests across all threads. Previously each thread slept on its own, so `max_workers=5` could send five requests per interval.
+- **Fragments no longer cause repeat crawls.** `page#one` and `page#two` are the same page. Host case and default ports are normalised too.
+- **Output is deterministic.** Results come out in discovery order regardless of thread count, and duplicate detection runs after the crawl. Previously duplicate detection depended on thread timing and could let near-duplicates through.
+- **Failures no longer discard output.** One unreachable host or malformed URL used to abort the crawl before anything was written. Output is now written for every page that succeeded.
+- **Blank lines in links files** no longer crash the run.
+- **Character encoding.** Pages are decoded from their declared charset (HTTP header or `<meta charset>`). Previously requests decoded `text/*` responses without a charset as ISO-8859-1, which mangled UTF-8 pages.
+- **`<base href>`** is honoured when resolving relative links and images.
+- **Non-text responses** (images, PDFs and similar) are no longer parsed as HTML when they are reached as child links.
+- **`follow_child_links`** respects `max_depth`, which is new in this release.
+
+### New options
+- `scrape_text` and `scrape_images` (keyword-only): `timeout`, `max_page_bytes`, `max_depth`, `block_private_addresses`, `user_agent`, `max_redirects`.
+- `scrape_images` (keyword-only): `max_image_bytes`.
+- `scrape_text`: `output_file=None` skips the plain-text file.
+- `pywebscrapr.__version__` exposes the package version.
+
+### Packaging, CI and tests
+- The version is defined once, in `pywebscrapr/__init__.py`, and `setup.py` reads it. The `{{VERSION_PLACEHOLDER}}` substitution is gone.
+- `bs4` is replaced by `beautifulsoup4` in `requirements.txt` and `setup.py`. Build with `python -m build`.
+- The workflow runs the offline test suite on Python 3.9, 3.12 and 3.13 for every push and pull request. Publishing to PyPI happens only for tags, only after the tests pass, and only if the tag matches `__version__`.
+- `tests.py` is rewritten: 39 offline tests run against local fixture servers, so no internet access is needed. Set `PYWEBSCRAPR_LIVE_TESTS=1` to also run a live check against example.com.
+- `.gitignore` covers `output/` and `*.egg-info/`.
+
## Changes in 0.1.7:
- Optimized `is_similar_regex` text similarity and duplicate checking by caching text normalization and word splitting, drastically improving duplicate text detection speed by up to 25x (~96% CPU time reduction).
@@ -45,17 +110,7 @@ pip install pywebscrapr
## Supported Python Versions
-PyWebScrapr supports the following Python versions:
-
-- Python 3.6
-- Python 3.7
-- Python 3.8
-- Python 3.9
-- Python 3.10
-- Python 3.11
-- Python 3.12/Later (Preferred)
-
-Please ensure that you have one of these Python versions installed before using PyWebScrapr. PyWebScrapr may not work as expected on lower versions of Python than the supported.
+PyWebScrapr supports Python 3.9 and later. Python 3.9, 3.12 and 3.13 are tested in CI.
## Features
@@ -92,6 +147,40 @@ links_array = ['https://example.com/image1.jpg', 'https://example.com/image2.png
scrape_images(links_file=links_file, links_array=links_array, save_folder='images')
```
+### Options: timeouts, limits and safety
+
+Every request has a timeout and a size cap by default. Override them with keyword arguments:
+
+```python
+from pywebscrapr import scrape_images, scrape_text
+
+scrape_text(
+ links_array=['https://example.com'],
+ output_file='output.txt',
+ follow_child_links=True,
+ max_links_to_follow=20, # child links only; seed links are always scraped
+ max_depth=2, # child-link depth (0 = seed pages only)
+ timeout=(5, 20), # (connect, read) seconds; None disables timeouts
+ max_page_bytes=5 * 1024**2, # largest page accepted
+ rate_limit=0.5, # minimum seconds between requests, shared by all threads
+)
+
+scrape_images(
+ links_array=['https://example.com/gallery'],
+ save_folder='images',
+ max_image_bytes=10 * 1024**2,
+ block_private_addresses=True, # refuse loopback, private and link-local targets
+ user_agent='MyCrawler/1.0 (+https://example.com/bot)',
+)
+```
+
+Skipped pages and images are logged. To see them, configure logging:
+
+```python
+import logging
+logging.basicConfig(level=logging.INFO) # logger name: 'pywebscrapr'
+```
+
## Contributing
Contributions are welcome! If you encounter any issues, have suggestions, or want to contribute to PyWebScrapr, please open an issue or submit a pull request on [GitHub](https://github.com/infinitode/pywebscrapr).
diff --git a/pywebscrapr/__init__.py b/pywebscrapr/__init__.py
index 3d4020f..43df701 100644
--- a/pywebscrapr/__init__.py
+++ b/pywebscrapr/__init__.py
@@ -1,175 +1,72 @@
-import os
+"""PyWebScrapr: scrape text, links and images from web pages.
+
+Failures on individual URLs are skipped and logged through the standard
+``logging`` module (logger name ``pywebscrapr``); they do not abort a run.
+"""
import csv
+import hashlib
+import ipaddress
import json
-import requests
+import logging
+import os
import re
+import socket
+import threading
import time
-from threading import Lock
-from concurrent.futures import ThreadPoolExecutor, as_completed
-from bs4 import BeautifulSoup, SoupStrainer
-from urllib.parse import urlparse, urljoin
-from PIL import Image
+from collections import namedtuple
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+from functools import lru_cache
from io import BytesIO
+from urllib.parse import unquote, urljoin, urlparse, urlunparse
-def _get_absolute_url(base_url, relative_url):
- """Helper function to get absolute URLs."""
- return urljoin(base_url, relative_url)
-
-def _download_image(url, save_path):
- """Helper function to download an image."""
- response = requests.get(url)
- with open(save_path, 'wb') as file:
- file.write(response.content)
-
-def _check_image_dimensions(img_source, min_width=None, min_height=None, max_width=None, max_height=None):
- """Helper function to check image dimensions, handling both URL and BytesIO sources."""
- try:
- if isinstance(img_source, str): # If img_source is a URL
- response = requests.get(img_source, stream=True)
- response.raise_for_status()
-
- # Check if the content type is an image
- if 'image' not in response.headers.get('Content-Type', ''):
- print(f"Skipped non-image content: {img_source}")
- return False
-
- # Check for minimal content length to avoid error pages
- content_length = response.headers.get('Content-Length')
- if content_length and int(content_length) < 500: # arbitrary small size for images
- print(f"Skipped due to small content size: {img_source}")
- return False
-
- img_data = BytesIO(response.content)
- img = Image.open(img_data)
- elif isinstance(img_source, BytesIO): # Already a BytesIO object
- img = Image.open(img_source)
- else:
- raise ValueError("img_source must be a URL (str) or BytesIO object.")
-
- img.verify() # Verify it is indeed an image
-
- # Re-open after verification as img.verify() closes the file
- img = Image.open(img_data if isinstance(img_source, str) else img_source)
- width, height = img.size
-
- # Check against provided constraints
- if (min_width and width < min_width) or (min_height and height < min_height):
- return False
- if (max_width and width > max_width) or (max_height and height > max_height):
- return False
-
- return True
- except (Image.UnidentifiedImageError, requests.exceptions.RequestException, ValueError) as e:
- print(f"Error checking image dimensions: {e}")
- return False
-
-def scrape_images(links_file=None, links_array=None, save_folder='images', min_width=None, min_height=None, max_width=None, max_height=None, follow_child_links=False, max_links_to_follow=None, print_progress=False, rate_limit=0, max_workers=5):
- """
- Scrape image content from the given links and save to specified output folder.
-
- Parameters:
- - links_file (str): Path to a file containing links, with each link on a new line.
- - links_array (list): List of links to scrape images from.
- - save_folder (str): Folder to save the scraped images.
- - min_width (int): Minimum width of images to include (optional).
- - min_height (int): Minimum height of images to include (optional).
- - max_width (int): Maximum width of images to include (optional).
- - max_height (int): Maximum height of images to include (optional).
- - follow_child_links (bool): If True, follow and scrape images from child links found on the page.
- - max_links_to_follow (int): Maximum number of links to follow when scraping child links.
- - print_progress (bool): If True, print progress updates to the console.
- - rate_limit (int): Seconds to wait between requests.
- - max_workers (int): Maximum number of threads to use.
+import requests
+from bs4 import BeautifulSoup, NavigableString, SoupStrainer, Tag
+from PIL import Image
- Example:
- ```python
- from pywebscrapr import scrape_images
+__version__ = '0.1.8'
- # Using links from a file and saving images to output_images folder.
- scrape_images(links_file='links.txt', save_folder='output_images', min_width=100, min_height=100, follow_child_links=True, max_links_to_follow=10)
- ```
- """
- def get_links_from_page(url):
- """Helper function to get all links from a page."""
- response = requests.get(url)
- soup = BeautifulSoup(response.text, 'html.parser')
- return [urljoin(url, a['href']) for a in soup.find_all('a', href=True)]
+logger = logging.getLogger('pywebscrapr')
- def is_same_domain(url1, url2):
- """Helper function to check if two URLs belong to the same domain."""
- return urlparse(url1).netloc == urlparse(url2).netloc
+DEFAULT_TIMEOUT = (10, 30) # (connect, read) seconds
+DEFAULT_MAX_PAGE_BYTES = 10 * 1024 * 1024
+DEFAULT_MAX_IMAGE_BYTES = 20 * 1024 * 1024
+DEFAULT_MAX_REDIRECTS = 10
+DEFAULT_USER_AGENT = 'PyWebScrapr/{} (+https://github.com/infinitode/pywebscrapr)'.format(__version__)
- if not os.path.exists(save_folder):
- os.makedirs(save_folder)
+_MIN_IMAGE_BYTES = 500 # responses smaller than this are treated as error pages
+_CHUNK_SIZE = 64 * 1024
+_REDIRECT_CODES = (301, 302, 303, 307, 308)
+_ELEMENT_MODES = ('text', 'content', 'unseen', 'links')
- links = []
- if links_file:
- with open(links_file, 'r') as file:
- links = file.read().splitlines()
- elif links_array:
- links = links_array
- else:
- raise ValueError("Either 'links_file' or 'links_array' must be provided.")
+# Text inside these elements is never emitted as page content.
+_SKIPPED_TEXT_TAGS = frozenset(['head', 'title', 'script', 'style', 'noscript', 'template'])
+# Text is grouped under its nearest block-level ancestor so that inline markup
+# does not split words and nested containers do not repeat text.
+_BLOCK_TAGS = frozenset([
+ 'address', 'article', 'aside', 'blockquote', 'body', 'caption', 'dd', 'details', 'div', 'dl',
+ 'dt', 'figcaption', 'figure', 'footer', 'form', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'header',
+ 'li', 'main', 'nav', 'ol', 'option', 'p', 'pre', 'section', 'summary', 'table', 'tbody', 'td',
+ 'tfoot', 'th', 'thead', 'tr', 'ul',
+])
+_FORMAT_EXTENSIONS = {'JPEG': '.jpg', 'MPO': '.jpg', 'TIFF': '.tif'}
+_RESERVED_NAMES = frozenset(['con', 'prn', 'aux', 'nul'] +
+ ['com{}'.format(i) for i in range(1, 10)] +
+ ['lpt{}'.format(i) for i in range(1, 10)])
- visited_links = set() # Track visited links to avoid loops
- strainer = SoupStrainer('img')
- links_followed = 0 # Track the number of links followed
- lock = Lock()
+_Fetched = namedtuple('_Fetched', ['url', 'content_type', 'body'])
- def scrape_page(link):
- """Helper function to scrape images from a single page."""
- nonlocal links_followed
- with lock:
- if link in visited_links or (max_links_to_follow and links_followed >= max_links_to_follow):
- return link, []
- visited_links.add(link)
- links_followed += 1
- if rate_limit > 0:
- time.sleep(rate_limit)
- response = requests.get(link)
- soup = BeautifulSoup(response.text, 'html.parser', parse_only=strainer)
+class _UnsafeURLError(Exception):
+ """Raised for URLs that use a disallowed scheme or resolve to a blocked address."""
- for img_tag in soup.find_all('img'):
- img_url = _get_absolute_url(link, img_tag.get('src'))
-
- # Skip data URLs
- if not img_url.startswith('data:'):
- img_name = os.path.basename(urlparse(img_url).path)
- save_path = os.path.join(save_folder, img_name)
-
- # Check image dimensions before downloading
- if _check_image_dimensions(img_url, min_width, min_height, max_width, max_height):
- _download_image(img_url, save_path)
- print(f"Downloaded: {img_url} -> {save_path}")
- else:
- print(f"Ignored due to size constraints: {img_url}")
-
- child_links_to_scrape = []
- if follow_child_links:
- child_links = get_links_from_page(link)
- for child_link in child_links:
- if is_same_domain(link, child_link):
- child_links_to_scrape.append(child_link)
- return link, child_links_to_scrape
-
- with ThreadPoolExecutor(max_workers=max_workers) as executor:
- futures = {executor.submit(scrape_page, link): link for link in links}
- scraped_count = 0
- while futures:
- for future in as_completed(list(futures)):
- link, child_links = future.result()
- scraped_count += 1
- if print_progress:
- print(f"Scraped {scraped_count} pages: {link}")
- del futures[future]
+class _ResponseTooLargeError(Exception):
+ """Raised when a response body exceeds the configured size limit."""
- if child_links:
- for child_link in child_links:
- if child_link not in visited_links:
- futures[executor.submit(scrape_page, child_link)] = child_link
+# ---------------------------------------------------------------------------
+# Public helpers (unchanged signatures)
+# ---------------------------------------------------------------------------
def normalize_text(text):
"""Normalize the text by converting to lowercase, removing extra whitespace and irrelevant patterns."""
@@ -179,183 +76,709 @@ def normalize_text(text):
text = re.sub(r'\W+', ' ', text) # Remove non-word characters
return text.strip()
-from functools import lru_cache
@lru_cache(maxsize=10000)
def _cached_normalize_words(text):
"""Helper to convert text string into normalized word frozenset with caching."""
return frozenset(normalize_text(text).split())
+
def _get_normalized_words(text):
"""Helper to convert text into normalized word frozenset, caching string results to avoid re-normalization overhead."""
if isinstance(text, (set, frozenset)):
return text
return _cached_normalize_words(text)
+
def is_similar_regex(new_text, seen_texts, threshold=0.8):
"""
- Check if the `new_text` is similar to any text in `seen_texts` based on regex similarity.
- Optimized using LRU caching to avoid redundant text normalization and splitting on repeated comparisons.
+ Check if the `new_text` is similar to any text in `seen_texts` based on word overlap.
+
+ Similarity is the number of shared words divided by the word count of the *larger*
+ of the two texts. Measuring against the larger text means a short item (for example a
+ navigation label such as "Home") cannot mark a long, unrelated paragraph as a duplicate.
"""
new_text_words = _get_normalized_words(new_text)
+ if not new_text_words:
+ return False
for seen_item in seen_texts:
seen_text_words = _get_normalized_words(seen_item)
+ if not seen_text_words:
+ continue # Skip empty text comparisons
- # Skip empty text comparisons
- if len(seen_text_words) == 0:
- continue
-
- # Calculate similarity ratio
- match_ratio = len(new_text_words & seen_text_words) / len(seen_text_words)
-
- # If match ratio exceeds threshold, consider as a duplicate
+ shared = len(new_text_words & seen_text_words)
+ match_ratio = shared / max(len(new_text_words), len(seen_text_words))
if match_ratio >= threshold:
return True
return False
+
+# ---------------------------------------------------------------------------
+# URL handling
+# ---------------------------------------------------------------------------
+
+def _get_absolute_url(base_url, relative_url):
+ """Helper function to get absolute URLs."""
+ return urljoin(base_url, relative_url)
+
+
+def _canonical_parts(url):
+ """Return (scheme, netloc, path, query) with host lower-cased and default ports removed."""
+ parts = urlparse(url)
+ scheme = parts.scheme.lower()
+ host = parts.hostname or ''
+ if ':' in host: # IPv6 literal
+ host = '[{}]'.format(host)
+ port = parts.port
+ default_port = {'http': 80, 'https': 443}.get(scheme)
+ netloc = host if port is None or port == default_port else '{}:{}'.format(host, port)
+ return scheme, netloc, parts.path or '/', parts.query
+
+
+def _canonical_url(url):
+ """Canonical form of a URL used for de-duplication (fragments are dropped)."""
+ scheme, netloc, path, query = _canonical_parts(url)
+ return urlunparse((scheme, netloc, path, '', query, ''))
+
+
+def _netloc(url):
+ return _canonical_parts(url)[1]
+
+
+def _to_ip(text):
+ ip = ipaddress.ip_address(text.split('%', 1)[0]) # strip IPv6 zone id
+ if ip.version == 6 and ip.ipv4_mapped is not None:
+ ip = ip.ipv4_mapped
+ return ip
+
+
+def _resolve(host, port):
+ try:
+ return [_to_ip(host)] # host is an IP literal
+ except ValueError:
+ pass
+ try:
+ infos = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)
+ except socket.gaierror as exc:
+ raise _UnsafeURLError('cannot resolve host {!r}: {}'.format(host, exc))
+ addresses = []
+ for info in infos:
+ try:
+ addresses.append(_to_ip(info[4][0]))
+ except ValueError:
+ continue
+ if not addresses:
+ raise _UnsafeURLError('host {!r} did not resolve to an IP address'.format(host))
+ return addresses
+
+
+def _check_url(url, block_private_addresses):
+ """Raise _UnsafeURLError if the URL is not an allowed http(s) target."""
+ parts = urlparse(url)
+ if parts.scheme not in ('http', 'https'):
+ raise _UnsafeURLError('unsupported URL scheme {!r}'.format(parts.scheme))
+ host = parts.hostname
+ if not host:
+ raise _UnsafeURLError('URL has no host name')
+ if block_private_addresses:
+ port = parts.port or (443 if parts.scheme == 'https' else 80)
+ for address in _resolve(host, port):
+ if not address.is_global:
+ raise _UnsafeURLError('{} resolves to a non-public address ({})'.format(host, address))
+
+
+# ---------------------------------------------------------------------------
+# HTTP layer
+# ---------------------------------------------------------------------------
+
+def _read_body(response, max_bytes):
+ """Read a response body, enforcing an optional byte limit."""
+ length = response.headers.get('Content-Length', '').strip()
+ if max_bytes and length.isdigit() and int(length) > max_bytes:
+ raise _ResponseTooLargeError('Content-Length {} exceeds the limit of {} bytes'.format(length, max_bytes))
+ chunks = []
+ total = 0
+ for chunk in response.iter_content(chunk_size=_CHUNK_SIZE):
+ if not chunk:
+ continue
+ total += len(chunk)
+ if max_bytes and total > max_bytes:
+ raise _ResponseTooLargeError('response exceeds the limit of {} bytes'.format(max_bytes))
+ chunks.append(chunk)
+ return b''.join(chunks)
+
+
+class _Fetcher(object):
+ """Thread-aware HTTP client shared by one scraping run."""
+
+ def __init__(self, timeout, user_agent, max_redirects, block_private_addresses, rate_limit):
+ self.timeout = timeout
+ self.user_agent = user_agent
+ self.max_redirects = max_redirects
+ self.block_private_addresses = block_private_addresses
+ self.rate_limit = rate_limit
+ self._local = threading.local()
+ self._sessions = []
+ self._sessions_lock = threading.Lock()
+ self._slot_lock = threading.Lock()
+ self._next_slot = 0.0
+
+ def _session(self):
+ # requests.Session is not guaranteed to be thread-safe, so each worker thread gets its own.
+ session = getattr(self._local, 'session', None)
+ if session is None:
+ session = requests.Session()
+ if self.user_agent:
+ session.headers['User-Agent'] = self.user_agent
+ with self._sessions_lock:
+ self._sessions.append(session)
+ self._local.session = session
+ return session
+
+ def _throttle(self):
+ """Global rate limit: requests are spaced `rate_limit` seconds apart across all threads."""
+ if not self.rate_limit:
+ return
+ with self._slot_lock:
+ now = time.monotonic()
+ slot = max(now, self._next_slot)
+ self._next_slot = slot + self.rate_limit
+ delay = slot - now
+ if delay > 0:
+ time.sleep(delay)
+
+ def get(self, url, max_bytes):
+ """GET a URL, following redirects manually so every hop is validated."""
+ current = url
+ for _ in range(self.max_redirects + 1):
+ _check_url(current, self.block_private_addresses)
+ self._throttle()
+ response = self._session().get(current, timeout=self.timeout, allow_redirects=False, stream=True)
+ try:
+ location = response.headers.get('Location')
+ if response.status_code in _REDIRECT_CODES and location:
+ current = urljoin(current, location)
+ continue
+ response.raise_for_status()
+ body = _read_body(response, max_bytes)
+ return _Fetched(current, response.headers.get('Content-Type', ''), body)
+ finally:
+ response.close()
+ raise requests.exceptions.TooManyRedirects('exceeded {} redirects'.format(self.max_redirects))
+
+ def close(self):
+ with self._sessions_lock:
+ for session in self._sessions:
+ session.close()
+ self._sessions = []
+
+
+def _charset_from_content_type(content_type):
+ match = re.search(r'charset=["\']?([A-Za-z0-9_.:-]+)', content_type or '', re.I)
+ return match.group(1) if match else None
+
+
+def _is_textual(content_type):
+ if not content_type:
+ return True
+ mime = content_type.split(';', 1)[0].strip().lower()
+ return mime.startswith('text/') or mime in ('application/xhtml+xml', 'application/xml')
+
+
+def _parse(body, content_type, strainer=None):
+ return BeautifulSoup(body, 'html.parser', from_encoding=_charset_from_content_type(content_type),
+ parse_only=strainer)
+
+
+def _base_url(soup, page_url):
+ base = soup.find('base', href=True)
+ if base is None:
+ return page_url
+ try:
+ return urljoin(page_url, base['href'].strip())
+ except ValueError:
+ return page_url
+
+
+def _child_urls(soup, base, origin_netloc):
+ """Yield same-domain child links (canonical form) found on a page."""
+ for anchor in soup.find_all('a', href=True):
+ href = anchor['href'].strip()
+ if not href:
+ continue
+ url = urljoin(base, href)
+ if urlparse(url).scheme not in ('http', 'https'):
+ continue
+ try:
+ key = _canonical_url(url)
+ except ValueError:
+ continue
+ if _netloc(key) == origin_netloc:
+ yield key
+
+
+# ---------------------------------------------------------------------------
+# Crawl engine
+# ---------------------------------------------------------------------------
+
+def _crawl(seeds, fetcher, process, *, strainer, follow_child_links, max_links_to_follow,
+ max_depth, max_workers, max_page_bytes, print_progress):
+ """
+ Fetch the seed URLs and, optionally, same-domain child links concurrently.
+
+ `process(fetched, soup, base_url)` runs in worker threads and returns the page's result.
+ Pages that fail are skipped and logged. Returns [(final_url, result), ...] in the order
+ the pages were discovered, so output is deterministic for a given input.
+ """
+ lock = threading.Lock()
+ seen = set()
+ state = {'seq': 0, 'children': 0, 'scraped': 0}
+ results = {}
+
+ def visit(url, depth, origin):
+ try:
+ fetched = fetcher.get(url, max_page_bytes)
+ if origin is not None and _netloc(fetched.url) != origin:
+ logger.warning('Skipped %s: child link redirected to a different domain (%s)', url, fetched.url)
+ return None
+ if not _is_textual(fetched.content_type):
+ logger.info('Skipped non-text content: %s (%s)', fetched.url, fetched.content_type)
+ return None
+ soup = _parse(fetched.body, fetched.content_type, strainer)
+ base = _base_url(soup, fetched.url)
+ result = process(fetched, soup, base)
+ children = []
+ if follow_child_links and (max_depth is None or depth < max_depth):
+ children = list(_child_urls(soup, base, _netloc(fetched.url)))
+ return fetched.url, result, children, depth
+ except Exception as exc: # one bad page must never abort the run
+ logger.warning('Skipped %s: %s', url, exc)
+ return None
+
+ def queue(executor, pending, url, depth, origin, is_child):
+ try:
+ key = _canonical_url(url)
+ except ValueError as exc:
+ logger.warning('Skipped invalid URL %r: %s', url, exc)
+ return
+ with lock:
+ if key in seen:
+ return
+ if is_child:
+ if max_links_to_follow is not None and state['children'] >= max_links_to_follow:
+ return
+ state['children'] += 1
+ seen.add(key)
+ seq = state['seq']
+ state['seq'] += 1
+ pending[executor.submit(visit, url, depth, origin)] = seq
+
+ pending = {}
+ executor = ThreadPoolExecutor(max_workers=max_workers)
+ try:
+ for url in seeds:
+ queue(executor, pending, url, 0, None, False)
+ while pending:
+ done, _ = wait(list(pending), return_when=FIRST_COMPLETED)
+ for future in done:
+ seq = pending.pop(future)
+ outcome = future.result() # visit() never raises
+ if outcome is None:
+ continue
+ final_url, result, children, depth = outcome
+ results[seq] = (final_url, result)
+ state['scraped'] += 1
+ if print_progress:
+ print('Scraped {} pages: {}'.format(state['scraped'], final_url))
+ origin = _netloc(final_url)
+ for child in children:
+ queue(executor, pending, child, depth + 1, origin, True)
+ finally:
+ for future in pending:
+ future.cancel()
+ executor.shutdown(wait=True)
+ fetcher.close()
+ return [results[seq] for seq in sorted(results)]
+
+
+# ---------------------------------------------------------------------------
+# Text extraction
+# ---------------------------------------------------------------------------
+
+def _block_for(node):
+ """Return (nearest block ancestor or None, skip flag) for a node."""
+ block = None
+ for parent in node.parents:
+ name = parent.name
+ if name in _SKIPPED_TEXT_TAGS:
+ return None, True
+ if block is None and name in _BLOCK_TAGS:
+ block = parent
+ return block, False
+
+
+def _extract_text_blocks(soup):
+ """
+ Return the visible text of each block-level element, in document order.
+
+ Each block's text is built from its inline pieces (so `Hello <b>world</b>` stays
+ "Hello world" and `Hel<b>lo</b>` stays "Hello"), and every piece of text is counted
+ once, under its nearest block ancestor. `<br>` becomes a line break.
+ """
+ order = []
+ parts = {}
+ for node in soup.descendants:
+ if isinstance(node, Tag):
+ if node.name != 'br':
+ continue
+ text = '\n'
+ elif type(node) is NavigableString: # excludes comments, doctypes, script and style strings
+ text = str(node)
+ else:
+ continue
+ block, skipped = _block_for(node)
+ if skipped:
+ continue
+ key = id(block) if block is not None else None
+ if key not in parts:
+ parts[key] = []
+ order.append(key)
+ parts[key].append(text)
+ return [''.join(parts[key]) for key in order]
+
+
+def _extract_items(soup, base, mode, remove_extra_whitespace):
+ """Extract the requested content type from a parsed page as a list of strings."""
+ if mode == 'text':
+ raw = _extract_text_blocks(soup)
+ elif mode == 'content':
+ raw = [meta.get('content') for meta in soup.find_all('meta')]
+ elif mode == 'unseen':
+ raw = [meta.get('content') for meta in soup.find_all('meta')]
+ raw += [script.get_text() for script in soup.find_all('script')]
+ else: # 'links'
+ raw = []
+ for element in soup.find_all(['a', 'img', 'video', 'audio', 'source']):
+ value = element.get('href') if element.name == 'a' else element.get('src')
+ if value and value.strip():
+ url = urljoin(base, value.strip())
+ if urlparse(url).scheme in ('http', 'https'):
+ raw.append(url)
+
+ items = []
+ for value in raw:
+ if not isinstance(value, str):
+ continue
+ text = ' '.join(value.split()) if remove_extra_whitespace else value.strip()
+ if text:
+ items.append(text)
+ return items
+
+
+def _csv_safe(value):
+ """Neutralise spreadsheet formula injection by prefixing risky cells with an apostrophe."""
+ if isinstance(value, str) and value[:1] in ('=', '+', '-', '@', '\t', '\r'):
+ return "'" + value
+ return value
+
+
+# ---------------------------------------------------------------------------
+# Image handling
+# ---------------------------------------------------------------------------
+
+def _safe_stem(url):
+ """Build a filesystem-safe base name from the URL path."""
+ path = unquote(urlparse(url).path)
+ name = re.split(r'[\\/]', path)[-1]
+ stem = os.path.splitext(name)[0]
+ stem = re.sub(r'[^A-Za-z0-9._-]+', '_', stem).strip('._-')[:100]
+ if not stem:
+ return 'image'
+ if stem.lower() in _RESERVED_NAMES:
+ return 'image_' + stem
+ return stem
+
+
+def _reserve_path(folder, url, image_format, used, used_lock):
+ """Pick a file path that no other image in this run has used (case-insensitive)."""
+ stem = _safe_stem(url)
+ extension = _FORMAT_EXTENSIONS.get(image_format.upper(), '.' + image_format.lower())
+ with used_lock:
+ name = stem + extension
+ count = 0
+ while name.lower() in used:
+ count += 1
+ digest = hashlib.sha1(url.encode('utf-8')).hexdigest()[:8]
+ suffix = digest if count == 1 else '{}-{}'.format(digest, count)
+ name = '{}-{}{}'.format(stem, suffix, extension)
+ used.add(name.lower())
+ return os.path.join(folder, name)
+
+
+def _inspect_image(body):
+ """Verify the bytes are a real image and return (format, width, height)."""
+ buffer = BytesIO(body)
+ with Image.open(buffer) as img:
+ img.verify() # raises on corrupt data
+ buffer.seek(0) # verify() invalidates the image, so reopen it
+ with Image.open(buffer) as img:
+ image_format, (width, height) = img.format, img.size
+ if not image_format:
+ raise ValueError('unrecognised image format')
+ return image_format, width, height
+
+
+def _within_bounds(width, height, min_width, min_height, max_width, max_height):
+ if (min_width and width < min_width) or (min_height and height < min_height):
+ return False
+ if (max_width and width > max_width) or (max_height and height > max_height):
+ return False
+ return True
+
+
+def _save_image(image_url, fetcher, folder, bounds, max_image_bytes, used, used_lock):
+ """Fetch, validate and save one image. Each image is downloaded exactly once."""
+ try:
+ fetched = fetcher.get(image_url, max_image_bytes)
+ mime = fetched.content_type.split(';', 1)[0].strip().lower()
+ if not mime.startswith('image/'):
+ logger.info('Skipped non-image content: %s (%s)', image_url, fetched.content_type)
+ return
+ if len(fetched.body) < _MIN_IMAGE_BYTES:
+ logger.info('Skipped due to small content size: %s', image_url)
+ return
+ image_format, width, height = _inspect_image(fetched.body)
+ if not _within_bounds(width, height, *bounds):
+ logger.info('Ignored due to size constraints: %s', image_url)
+ return
+ path = _reserve_path(folder, image_url, image_format, used, used_lock)
+ with open(path, 'wb') as file:
+ file.write(fetched.body)
+ logger.info('Downloaded: %s -> %s', image_url, path)
+ except Exception as exc: # corrupt, oversized, unreachable: skip and continue
+ logger.warning('Skipped image %s: %s', image_url, exc)
+
+
+# ---------------------------------------------------------------------------
+# Input and option validation
+# ---------------------------------------------------------------------------
+
+def _check_optional_int(name, value, minimum):
+ if value is None:
+ return
+ if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
+ raise ValueError('{} must be None or an integer >= {}'.format(name, minimum))
+
+
+def _check_timeout(timeout):
+ def valid(value):
+ return value is None or (isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0)
+
+ if isinstance(timeout, tuple):
+ if len(timeout) != 2 or not all(valid(part) for part in timeout):
+ raise ValueError('timeout tuple must be (connect, read) with positive numbers or None')
+ elif not valid(timeout):
+ raise ValueError('timeout must be a positive number, a (connect, read) tuple, or None')
+
+
+def _validate_options(max_workers, rate_limit, timeout, max_links_to_follow, max_depth,
+ max_redirects, byte_limits):
+ if isinstance(max_workers, bool) or not isinstance(max_workers, int) or max_workers < 1:
+ raise ValueError('max_workers must be an integer >= 1')
+ if not isinstance(rate_limit, (int, float)) or isinstance(rate_limit, bool) or rate_limit < 0:
+ raise ValueError('rate_limit must be a number of seconds >= 0')
+ _check_timeout(timeout)
+ _check_optional_int('max_links_to_follow', max_links_to_follow, 0)
+ _check_optional_int('max_depth', max_depth, 0)
+ _check_optional_int('max_redirects', max_redirects, 0)
+ for name, value in byte_limits.items():
+ _check_optional_int(name, value, 1)
+
+
+def _load_links(links_file, links_array):
+ """Return the non-blank links from a file or list. Non-string entries are skipped and logged."""
+ if links_file:
+ with open(links_file, 'r', encoding='utf-8-sig') as file:
+ raw = [line for line in file]
+ elif links_array:
+ raw = list(links_array)
+ else:
+ raise ValueError("Either 'links_file' or 'links_array' must be provided.")
+
+ links = []
+ for entry in raw:
+ if not isinstance(entry, str):
+ logger.warning('Skipped link that is not a string: %r', entry)
+ continue
+ entry = entry.strip()
+ if entry:
+ links.append(entry)
+ return links
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def scrape_images(links_file=None, links_array=None, save_folder='images', min_width=None, min_height=None,
+ max_width=None, max_height=None, follow_child_links=False, max_links_to_follow=None,
+ print_progress=False, rate_limit=0, max_workers=5, *, timeout=DEFAULT_TIMEOUT,
+ max_page_bytes=DEFAULT_MAX_PAGE_BYTES, max_image_bytes=DEFAULT_MAX_IMAGE_BYTES,
+ max_depth=None, block_private_addresses=False, user_agent=DEFAULT_USER_AGENT,
+ max_redirects=DEFAULT_MAX_REDIRECTS):
+ """
+ Scrape image content from the given links and save to specified output folder.
+
+ Parameters:
+ - links_file (str): Path to a file containing links, with each link on a new line.
+ - links_array (list): List of links to scrape images from.
+ - save_folder (str): Folder to save the scraped images.
+ - min_width (int): Minimum width of images to include (optional).
+ - min_height (int): Minimum height of images to include (optional).
+ - max_width (int): Maximum width of images to include (optional).
+ - max_height (int): Maximum height of images to include (optional).
+ - follow_child_links (bool): If True, follow and scrape images from child links found on the page.
+ - max_links_to_follow (int): Maximum number of child links to follow (seed links are always scraped).
+ - print_progress (bool): If True, print progress updates to the console.
+ - rate_limit (float): Minimum seconds between requests, shared by all threads.
+ - max_workers (int): Maximum number of threads to use.
+ - timeout (float | tuple | None): Request timeout in seconds, or (connect, read). Default (10, 30).
+ None disables timeouts (not recommended).
+ - max_page_bytes (int | None): Largest HTML page accepted, in bytes. Default 10 MB.
+ - max_image_bytes (int | None): Largest image accepted, in bytes. Default 20 MB.
+ - max_depth (int | None): Maximum child-link depth (0 = seed pages only). None = unlimited.
+ - block_private_addresses (bool): If True, refuse URLs that resolve to private, loopback,
+ link-local or otherwise non-public addresses. Default False (allowed).
+ - user_agent (str | None): User-Agent header to send. None keeps the requests default.
+ - max_redirects (int): Maximum redirects followed per request.
+
+ Failed or rejected images and pages are skipped and logged to the 'pywebscrapr' logger.
+
+ Example:
+ ```python
+ from pywebscrapr import scrape_images
+
+ # Using links from a file and saving images to output_images folder.
+ scrape_images(links_file='links.txt', save_folder='output_images', min_width=100, min_height=100, follow_child_links=True, max_links_to_follow=10)
+ ```
+ """
+ _validate_options(max_workers, rate_limit, timeout, max_links_to_follow, max_depth, max_redirects,
+ {'max_page_bytes': max_page_bytes, 'max_image_bytes': max_image_bytes})
+ seeds = _load_links(links_file, links_array)
+ os.makedirs(save_folder, exist_ok=True)
+
+ fetcher = _Fetcher(timeout, user_agent, max_redirects, block_private_addresses, rate_limit)
+ used_names = set()
+ used_lock = threading.Lock()
+ bounds = (min_width, min_height, max_width, max_height)
+
+ def process(fetched, soup, base):
+ for img_tag in soup.find_all('img'):
+ src = (img_tag.get('src') or '').strip()
+ if not src or src.lower().startswith('data:'):
+ continue
+ _save_image(urljoin(base, src), fetcher, save_folder, bounds, max_image_bytes,
+ used_names, used_lock)
+ return None
+
+ _crawl(seeds, fetcher, process, strainer=SoupStrainer(['a', 'img', 'base']),
+ follow_child_links=follow_child_links, max_links_to_follow=max_links_to_follow,
+ max_depth=max_depth, max_workers=max_workers, max_page_bytes=max_page_bytes,
+ print_progress=print_progress)
+
+
def scrape_text(links_file=None, links_array=None, output_file='output.txt', csv_output_file=None, json_output_file=None,
remove_extra_whitespace=True, remove_duplicates=False, similarity_threshold=0.8,
- elements_to_scrape='text', follow_child_links=False, max_links_to_follow=None, print_progress=False, rate_limit=0, max_workers=5):
+ elements_to_scrape='text', follow_child_links=False, max_links_to_follow=None, print_progress=False,
+ rate_limit=0, max_workers=5, *, timeout=DEFAULT_TIMEOUT, max_page_bytes=DEFAULT_MAX_PAGE_BYTES,
+ max_depth=None, block_private_addresses=False, user_agent=DEFAULT_USER_AGENT,
+ max_redirects=DEFAULT_MAX_REDIRECTS):
"""
Scrape content from the given links and save to specified output file(s).
Parameters:
- links_file (str): Path to a file containing links, with each link on a new line.
- links_array (list): List of links to scrape content from.
- - output_file (str): File to save the scraped content.
+ - output_file (str | None): File to save the scraped content as plain text. None skips it.
- csv_output_file (str): File to save the URL and scraped information in CSV format.
- json_output_file (str): File to save the URL and scraped information in JSON format.
- - remove_extra_whitespace (bool): If True, remove extra whitespace and empty lines from the output.
+ - remove_extra_whitespace (bool): If True, collapse whitespace and drop empty lines from the output.
- remove_duplicates (bool): If True, remove duplicate or highly similar paragraphs.
- - similarity_threshold (float): Similarity percentage (0-1) above which paragraphs are considered duplicates.
+ - similarity_threshold (float): Similarity (0-1) at or above which paragraphs are considered duplicates.
+ Similarity is measured against the larger of the two paragraphs.
- elements_to_scrape (str): Type of content to scrape. Options are:
- 'text' (default) - Scrape visible textual content.
- 'content' - Scrape the [content](http://_vscodecontentref_/3) attribute of meta tags.
- 'unseen' - Scrape hidden or non-visible elements (e.g., meta tags, script data).
- 'links' - Scrape [href](http://_vscodecontentref_/4) or `src` attributes of anchor and media elements.
- - follow_child_links (bool): If True, follow and scrape content from child links found on the page.
- - max_links_to_follow (int): Maximum number of links to follow when scraping child links.
+ 'text' (default) - Scrape visible textual content, one block element per line.
+ 'content' - Scrape the content attribute of meta tags.
+ 'unseen' - Scrape hidden or non-visible elements (meta tag content and script data).
+ 'links' - Scrape absolute URLs from href and src attributes of anchor and media elements.
+ - follow_child_links (bool): If True, follow and scrape content from same-domain child links.
+ - max_links_to_follow (int | None): Maximum number of child links to follow (seed links are always scraped).
- print_progress (bool): If True, print progress updates to the console.
- - rate_limit (int): Seconds to wait between requests.
+ - rate_limit (float): Minimum seconds between requests, shared by all threads.
- max_workers (int): Maximum number of threads to use.
+ - timeout (float | tuple | None): Request timeout in seconds, or (connect, read). Default (10, 30).
+ None disables timeouts (not recommended).
+ - max_page_bytes (int | None): Largest page accepted, in bytes. Default 10 MB.
+ - max_depth (int | None): Maximum child-link depth (0 = seed pages only). None = unlimited.
+ - block_private_addresses (bool): If True, refuse URLs that resolve to private, loopback,
+ link-local or otherwise non-public addresses. Default False (allowed).
+ - user_agent (str | None): User-Agent header to send. None keeps the requests default.
+ - max_redirects (int): Maximum redirects followed per request.
+
+ Failed or rejected pages are skipped and logged to the 'pywebscrapr' logger.
Example:
```python
scrape_text(links_array=['https://example.com'], similarity_threshold=0.9, follow_child_links=True, max_links_to_follow=10)
```
"""
- def get_links_from_page(url):
- """Helper function to get all links from a page."""
- response = requests.get(url)
- soup = BeautifulSoup(response.text, 'html.parser')
- return [urljoin(url, a['href']) for a in soup.find_all('a', href=True)]
-
- def is_same_domain(url1, url2):
- """Helper function to check if two URLs belong to the same domain."""
- return urlparse(url1).netloc == urlparse(url2).netloc
-
- links = []
- if links_file:
- with open(links_file, 'r') as file:
- links = file.read().splitlines()
- elif links_array:
- links = links_array
- else:
- raise ValueError("Either 'links_file' or 'links_array' must be provided.")
-
+ if elements_to_scrape not in _ELEMENT_MODES:
+ raise ValueError('elements_to_scrape must be one of {}'.format(', '.join(_ELEMENT_MODES)))
+ if not isinstance(similarity_threshold, (int, float)) or not 0 < similarity_threshold <= 1:
+ raise ValueError('similarity_threshold must be a number greater than 0 and at most 1')
+ _validate_options(max_workers, rate_limit, timeout, max_links_to_follow, max_depth, max_redirects,
+ {'max_page_bytes': max_page_bytes})
+ seeds = _load_links(links_file, links_array)
+
+ fetcher = _Fetcher(timeout, user_agent, max_redirects, block_private_addresses, rate_limit)
+
+ def process(fetched, soup, base):
+ return _extract_items(soup, base, elements_to_scrape, remove_extra_whitespace)
+
+ results = _crawl(seeds, fetcher, process, strainer=None,
+ follow_child_links=follow_child_links, max_links_to_follow=max_links_to_follow,
+ max_depth=max_depth, max_workers=max_workers, max_page_bytes=max_page_bytes,
+ print_progress=print_progress)
+
+ # De-duplicate in discovery order, after the crawl, so results do not depend on thread timing.
+ seen_texts = []
all_content = []
- csv_data = []
- json_data = []
- seen_texts = [] # Store unique paragraphs
- visited_links = set() # Track visited links to avoid loops
- strainer = SoupStrainer(['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'p', 'a', 'code', 'span', 'nav',
- 'footer', 'header', 'table', 'td', 'ul', 'ol', 'div', 'meta'])
- links_followed = 0 # Track the number of links followed
- lock = Lock()
-
- def scrape_page(link):
- """Helper function to scrape content from a single page."""
- nonlocal links_followed
- with lock:
- if link in visited_links or (max_links_to_follow and links_followed >= max_links_to_follow):
- return link, []
- visited_links.add(link)
- links_followed += 1
-
- if rate_limit > 0:
- time.sleep(rate_limit)
- response = requests.get(link)
- soup = BeautifulSoup(response.text, 'html.parser', parse_only=strainer)
-
- page_content = []
-
- for element in soup.find_all(lambda tag: tag.name not in ['script', 'style']):
- if elements_to_scrape == 'text':
- content = element.get_text(strip=remove_extra_whitespace) if remove_extra_whitespace else element.get_text()
- elif elements_to_scrape == 'content' and element.has_attr('content'):
- content = element['content']
- elif elements_to_scrape == 'unseen' and element.name in ['meta', 'script']:
- content = element.get('content', element.get_text(strip=True))
- elif elements_to_scrape == 'links' and element.name in ['a', 'img', 'video']:
- content = element.get('href') or element.get('src')
- else:
- continue # Skip unsupported elements for the chosen scrape type
-
- if content:
- if remove_duplicates:
- # Check if content is similar to existing unique paragraphs using regex-based similarity
- if not is_similar_regex(content, seen_texts, similarity_threshold):
- seen_texts.append(content)
- page_content.append(content)
- else:
- page_content.append(content)
-
- if page_content:
- with lock:
- all_content.append("\n".join(page_content))
- if csv_output_file or json_output_file:
- csv_data.append({'URL': link, 'Content': "\n".join(page_content)})
- json_data.append({'URL': link, 'Content': "\n".join(page_content)})
-
- child_links_to_scrape = []
- if follow_child_links:
- child_links = get_links_from_page(link)
- for child_link in child_links:
- if is_same_domain(link, child_link):
- child_links_to_scrape.append(child_link)
- return link, child_links_to_scrape
-
- with ThreadPoolExecutor(max_workers=max_workers) as executor:
- futures = {executor.submit(scrape_page, link): link for link in links}
- scraped_count = 0
- while futures:
- for future in as_completed(list(futures)):
- link, child_links = future.result()
- scraped_count += 1
- if print_progress:
- print(f"Scraped {scraped_count} pages: {link}")
-
- del futures[future]
-
- if child_links:
- for child_link in child_links:
- if child_link not in visited_links:
- futures[executor.submit(scrape_page, child_link)] = child_link
-
- # Save content to output file
- with open(output_file, 'w', encoding='utf-8') as text_file:
- text_file.write("\n\n".join(all_content).rstrip())
+ records = []
+ for final_url, items in results:
+ kept = []
+ for item in items or []:
+ if remove_duplicates:
+ if is_similar_regex(item, seen_texts, similarity_threshold):
+ continue
+ seen_texts.append(item)
+ kept.append(item)
+ if kept:
+ content = "\n".join(kept)
+ all_content.append(content)
+ records.append({'URL': final_url, 'Content': content})
+
+ if output_file:
+ with open(output_file, 'w', encoding='utf-8') as text_file:
+ text_file.write("\n\n".join(all_content).rstrip())
- # Save CSV data to CSV file
if csv_output_file:
with open(csv_output_file, 'w', newline='', encoding='utf-8') as csv_file:
- fieldnames = ['URL', 'Content']
- writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+ writer = csv.DictWriter(csv_file, fieldnames=['URL', 'Content'])
writer.writeheader()
- writer.writerows(csv_data)
+ for record in records:
+ writer.writerow({key: _csv_safe(value) for key, value in record.items()})
- # Save JSON data to JSON file
if json_output_file:
with open(json_output_file, 'w', encoding='utf-8') as json_file:
- json.dump(json_data, json_file, ensure_ascii=False, indent=4)
\ No newline at end of file
+ json.dump(records, json_file, ensure_ascii=False, indent=4)
diff --git a/requirements.txt b/requirements.txt
index f7ce5d8..b78296c 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
-requests
-bs4
-pillow
\ No newline at end of file
+requests>=2.32.4
+beautifulsoup4>=4.12
+pillow>=10.3.0
diff --git a/setup.py b/setup.py
index deb4786..e54cb42 100644
--- a/setup.py
+++ b/setup.py
@@ -1,35 +1,41 @@
-from setuptools import setup, find_packages
+import os
+import re
-with open("README.md", "r", encoding="utf-8") as fh:
+from setuptools import find_packages, setup
+
+here = os.path.abspath(os.path.dirname(__file__))
+
+with open(os.path.join(here, 'README.md'), 'r', encoding='utf-8') as fh:
long_description = fh.read()
+# The version lives in the package so that it is defined in exactly one place.
+with open(os.path.join(here, 'pywebscrapr', '__init__.py'), 'r', encoding='utf-8') as fh:
+ version = re.search(r"^__version__\s*=\s*['\"]([^'\"]+)['\"]", fh.read(), re.M).group(1)
+
setup(
name='pywebscrapr',
- version='{{VERSION_PLACEHOLDER}}',
+ version=version,
author='Infinitode Pty Ltd',
author_email='infinitode.ltd@gmail.com',
description='An open-source Python library for web scraping tasks. Includes support for both image scraping and text scraping.',
long_description=long_description,
long_description_content_type='text/markdown',
url='https://github.com/infinitode/pywebscrapr',
- packages=find_packages(),
+ packages=find_packages(include=['pywebscrapr', 'pywebscrapr.*']),
install_requires=[
- 'requests',
- 'bs4',
- 'pillow'
+ 'requests>=2.32.4',
+ 'beautifulsoup4>=4.12',
+ 'pillow>=10.3.0',
],
classifiers=[
'Development Status :: 5 - Production/Stable',
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
- 'Programming Language :: Python :: 3.6',
- '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',
],
- python_requires='>=3.6',
+ python_requires='>=3.9',
)
diff --git a/tests.py b/tests.py
index 3d3d879..b3eac3c 100644
--- a/tests.py
+++ b/tests.py
@@ -1,15 +1,505 @@
+"""Test suite for PyWebScrapr.
+
+The tests run offline: they start small HTTP servers on 127.0.0.1 and scrape those.
+Set PYWEBSCRAPR_LIVE_TESTS=1 to also run a single test against a real website.
+
+Run with: python -m unittest -v tests
+"""
+import io
+import json
+import logging
+import os
+import shutil
+import struct
+import tempfile
+import threading
+import time
import unittest
-from pywebscrapr import scrape_images, scrape_text
+import zlib
+from collections import Counter
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlsplit
+
+from bs4 import BeautifulSoup
+from PIL import Image
+
+from pywebscrapr import (_check_url, _csv_safe, _extract_text_blocks, _safe_stem, _UnsafeURLError,
+ is_similar_regex, scrape_images, scrape_text)
+
+# Skipped URLs are logged as warnings; keep the test output quiet unless a test asserts on them.
+logging.getLogger('pywebscrapr').setLevel(logging.CRITICAL)
+
+
+# ---------------------------------------------------------------------------
+# Fixture helpers
+# ---------------------------------------------------------------------------
+
+def png_bytes(width=300, height=300):
+ """A real PNG. Noise keeps the file above the library's minimum image size."""
+ buffer = io.BytesIO()
+ Image.effect_noise((width, height), 64).convert('RGB').save(buffer, 'PNG')
+ return buffer.getvalue()
+
+
+def bomb_png(side=20000):
+ """A PNG whose header declares side x side pixels: PIL rejects it as a decompression bomb."""
+ def chunk(kind, data):
+ return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data) & 0xffffffff)
+
+ header = struct.pack('>IIBBBBB', side, side, 8, 2, 0, 0, 0)
+ padding = chunk(b'tEXt', b'comment\x00' + b'x' * 1000) # keep the file above the minimum size
+ return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', header) + padding +
+ chunk(b'IDAT', zlib.compress(b'\x00')) + chunk(b'IEND', b''))
+
+
+class FixtureSite(object):
+ """A tiny HTTP server on 127.0.0.1 with scripted routes.
+
+ Route values:
+ str / bytes -> 200 text/html
+ ('image', bytes) -> 200 image/png
+ ('status', code, body) -> that status with text/html
+ ('redirect', location) -> 302 to location
+ ('sleep', seconds, body) -> waits, then 200 text/html
+ ('raw', content_type, bytes) -> 200 with the given content type
+ """
+
+ def __init__(self, routes=None):
+ self.routes = dict(routes or {})
+ self.hits = Counter()
+ self._lock = threading.Lock()
+ site = self
+
+ class Handler(BaseHTTPRequestHandler):
+ def log_message(self, *args):
+ pass
+
+ def do_GET(self):
+ with site._lock:
+ site.hits[self.path] += 1
+ spec = site.routes.get(self.path)
+ if spec is None:
+ spec = site.routes.get(urlsplit(self.path).path)
+ if spec is None:
+ return self._send(404, 'text/html', b'<html><body>not found</body></html>')
+ if isinstance(spec, str):
+ spec = spec.encode('utf-8')
+ if isinstance(spec, bytes):
+ return self._send(200, 'text/html; charset=utf-8', spec)
+ kind = spec[0]
+ if kind == 'image':
+ return self._send(200, 'image/png', spec[1])
+ if kind == 'status':
+ body = spec[2].encode('utf-8') if isinstance(spec[2], str) else spec[2]
+ return self._send(spec[1], 'text/html; charset=utf-8', body)
+ if kind == 'redirect':
+ self.send_response(302)
+ self.send_header('Location', spec[1])
+ self.send_header('Content-Length', '0')
+ self.end_headers()
+ return None
+ if kind == 'sleep':
+ time.sleep(spec[1])
+ body = spec[2].encode('utf-8') if isinstance(spec[2], str) else spec[2]
+ return self._send(200, 'text/html; charset=utf-8', body)
+ if kind == 'raw':
+ return self._send(200, spec[1], spec[2])
+ raise AssertionError('bad route spec {!r}'.format(spec))
+
+ def _send(self, code, content_type, body):
+ self.send_response(code)
+ self.send_header('Content-Type', content_type)
+ self.send_header('Content-Length', str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ self._server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+ self._server.daemon_threads = True
+ self.base = 'http://127.0.0.1:{}'.format(self._server.server_address[1])
+ self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
+ self._thread.start()
+
+ def url(self, path):
+ return self.base + path
+
+ def close(self):
+ self._server.shutdown()
+ self._server.server_close()
+
+
+class TempDirTestCase(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp(prefix='pywebscrapr-test-')
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp, ignore_errors=True)
+
+ def path(self, name):
+ return os.path.join(self.tmp, name)
+
+ def read(self, name):
+ with open(self.path(name), encoding='utf-8') as handle:
+ return handle.read()
+
+
+class SiteTestCase(TempDirTestCase):
+ routes = {}
+
+ def setUp(self):
+ super(SiteTestCase, self).setUp()
+ self.site = FixtureSite(self.routes)
+
+ def tearDown(self):
+ self.site.close()
+ super(SiteTestCase, self).tearDown()
+
+
+# ---------------------------------------------------------------------------
+# Pure logic
+# ---------------------------------------------------------------------------
+
+class SimilarityTests(unittest.TestCase):
+ def test_short_seen_text_does_not_swallow_long_paragraph(self):
+ paragraph = ('Welcome home. Our company has been building reliable software for twenty years '
+ 'and we love it.')
+ self.assertFalse(is_similar_regex(paragraph, ['Home'], 0.8))
+
+ def test_near_duplicates_are_detected(self):
+ original = 'the quick brown fox jumps over the lazy dog today'
+ self.assertTrue(is_similar_regex(original, ['the quick brown fox jumps over the lazy dog'], 0.8))
+
+ def test_unrelated_texts_are_not_similar(self):
+ self.assertFalse(is_similar_regex('apples and oranges', ['cars and trucks'], 0.8))
+
+ def test_empty_inputs(self):
+ self.assertFalse(is_similar_regex('', ['anything here'], 0.8))
+ self.assertFalse(is_similar_regex('something here', [''], 0.8))
+
+
+class TextExtractionTests(unittest.TestCase):
+ def blocks(self, html):
+ return _extract_text_blocks(BeautifulSoup(html, 'html.parser'))
+
+ def test_inline_markup_does_not_glue_words(self):
+ self.assertEqual(self.blocks('<p>Hello <b>world</b>, this is <i>great</i> news</p>'),
+ ['Hello world, this is great news'])
+
+ def test_nested_containers_do_not_repeat_text(self):
+ self.assertEqual(self.blocks('<div><div><p>Only once</p></div></div>'), ['Only once'])
+
+ def test_head_script_and_comments_are_excluded(self):
+ html = ('<html><head><title>Title'
+ 'Visible