Scrapling
Scrapling là một thư viện Python mạnh mẽ, linh hoạt, hiệu suất cao và không thể phát hiện, được thiết kế để giúp việc thu thập dữ liệu web trở nên đơn giản và dễ dàng. Đây là thư viện thu thập dữ liệu thích ứng đầu tiên có khả năng học hỏi từ các thay đổi của trang web và phát triển cùng với chúng. Trong khi các thư viện khác bị lỗi khi cấu trúc trang web cập nhật, Scrapling tự động định vị lại các phần tử và giữ cho các trình thu thập của bạn hoạt động trơn tru.
Các tính năng chính:
- Công nghệ thu thập dữ liệu thích ứng – Thư viện đầu tiên học hỏi từ các thay đổi của trang web và tự động phát triển. Khi cấu trúc trang web cập nhật, Scrapling tự động định vị lại các phần tử để đảm bảo hoạt động liên tục.
- Giả mạo dấu vân tay trình duyệt – Hỗ trợ khớp dấu vân tay TLS và mô phỏng tiêu đề trình duyệt thực.
- Khả năng thu thập dữ liệu ẩn danh –
StealthyFetchercó thể vượt qua các hệ thống chống bot nâng cao như Cloudflare Turnstile. - Hỗ trợ phiên duy trì – Cung cấp nhiều loại phiên, bao gồm
FetcherSession,DynamicSessionvàStealthySession, để thu thập dữ liệu đáng tin cậy và hiệu quả.
Tìm hiểu thêm trong [tài liệu chính thức].
Tại sao nên kết hợp Scrapeless và Scrapling?
Scrapling nổi bật về khả năng trích xuất dữ liệu web hiệu suất cao, hỗ trợ thu thập dữ liệu thích ứng và tích hợp AI. Nó đi kèm với nhiều lớp Fetcher tích hợp — Fetcher, DynamicFetcher và StealthyFetcher — để xử lý các tình huống khác nhau.
Tuy nhiên, khi đối mặt với các cơ chế chống bot nâng cao hoặc thu thập dữ liệu đồng thời quy mô lớn, một số thách thức vẫn có thể phát sinh, chẳng hạn như:
- Các trình duyệt cục bộ dễ dàng bị chặn bởi Cloudflare, AWS WAF hoặc reCAPTCHA.
- Tiêu thụ tài nguyên trình duyệt cao và hiệu suất hạn chế trong quá trình thu thập dữ liệu đồng thời lớn.
- Mặc dù
StealthyFetcherbao gồm các khả năng ẩn danh, các tình huống chống bot cực đoan vẫn yêu cầu hỗ trợ cơ sở hạ tầng mạnh mẽ hơn. - Quy trình gỡ lỗi phức tạp khiến việc xác định nguyên nhân gốc rễ của các lỗi thu thập dữ liệu trở nên khó khăn.
Trình duyệt Scrapeless Cloud giải quyết hoàn hảo những vấn đề này:
- Vượt qua chống bot chỉ với một cú nhấp: Tự động xử lý reCAPTCHA, Cloudflare Turnstile/Challenge, AWS WAF và các xác minh khác. Kết hợp với khả năng trích xuất thích ứng của Scrapling, nó cải thiện đáng kể tỷ lệ thành công.
- Khả năng mở rộng đồng thời không giới hạn: Mỗi tác vụ có thể khởi chạy 50–1000+ phiên trình duyệt trong vòng vài giây, loại bỏ các nút thắt hiệu suất cục bộ và tối đa hóa tiềm năng hiệu suất cao của Scrapling.
- Giảm chi phí từ 40–80%: So với các giải pháp đám mây tương tự, Scrapeless chỉ tốn 20–60% tổng chi phí và hỗ trợ thanh toán theo mức sử dụng — giúp nó trở nên phải chăng ngay cả đối với các dự án nhỏ.
- Công cụ gỡ lỗi trực quan: Với các tính năng Session Replay và Live URL, bạn có thể theo dõi quá trình thực thi của Scrapling trong thời gian thực, nhanh chóng xác định các lỗi thu thập dữ liệu và giảm chi phí gỡ lỗi.
- Tích hợp linh hoạt:
DynamicFetchervàPlayWrightFetcher(được xây dựng trên Playwright) của Scrapling có thể dễ dàng kết nối với Scrapeless Cloud Browser thông qua cấu hình — không cần viết lại logic hiện có. - Các nút dịch vụ Edge: Với các trung tâm dữ liệu toàn cầu, Scrapeless đạt được tốc độ khởi động và độ ổn định nhanh hơn 2–3 lần so với các trình duyệt đám mây khác, cung cấp hơn 90 triệu IP dân cư đáng tin cậy ở hơn 195 quốc gia để tăng tốc độ thực thi của Scrapling.
- Môi trường biệt lập & Phiên duy trì: Mỗi hồ sơ Scrapeless chạy trong một môi trường biệt lập với hỗ trợ đăng nhập liên tục, ngăn chặn sự can thiệp giữa các phiên và đảm bảo sự ổn định trong thu thập dữ liệu quy mô lớn.
- Cấu hình dấu vân tay linh hoạt: Scrapeless có thể tạo ngẫu nhiên hoặc tùy chỉnh hoàn toàn dấu vân tay trình duyệt. Khi kết hợp với
StealthyFetchercủa Scrapling, nó tiếp tục giảm rủi ro phát hiện và tăng đáng kể tỷ lệ thành công của việc thu thập dữ liệu.
Bắt đầu
Đăng nhập vào Scrapeless và lấy API Key của bạn.
Điều kiện tiên quyết
- Python 3.10+
- Tài khoản Scrapeless đã đăng ký với API Key hợp lệ
- Scrapling đã cài đặt (hoặc sử dụng image Docker chính thức)
pip install scrapling
# If you need dynamic or stealth fetchers:
pip install "scrapling[fetchers]"
# Install browser dependencies
scrapling install
Hoặc sử dụng image Docker chính thức:
docker pull pyd4vinci/scrapling
# or
docker pull ghcr.io/d4vinci/scrapling:latest
Khởi động nhanh
Dưới đây là một ví dụ đơn giản: sử dụng DynamicSession (được cung cấp bởi Scrapling) để kết nối với Scrapeless Cloud Browser thông qua điểm cuối WebSocket của nó, tìm nạp một trang và in phản hồi.
from urllib.parse import urlencode
from scrapling.fetchers import DynamicSession
# Configure your browser session
config = {
"token": "YOUR_API_KEY",
"sessionName": "scrapling-session",
"sessionTTL": "300", # 5 minutes
"proxyCountry": "ANY",
"sessionRecording": "false",
}
# Build WebSocket URL
ws_endpoint = f"wss://browser.scrapeless.com/api/v2/browser?{urlencode(config)}"
print('Connecting to Scrapeless...')
with DynamicSession(cdp_url=ws_endpoint, disable_resources=True) as s:
print("Connected!")
page = s.fetch("https://httpbin.org/headers", network_idle=True)
print(f"Page loaded, content length: {len(page.body)}")
print(page.json())
Lưu ý: Scrapeless Cloud Browser hỗ trợ các tùy chọn nâng cao như cấu hình proxy, dấu vân tay tùy chỉnh và công cụ giải CAPTCHA.
Tham khảo Tài liệu trình duyệt Scrapeless để biết thêm chi tiết.
Các trường hợp sử dụng phổ biến (với các ví dụ đầy đủ)
Trước khi bắt đầu, hãy đảm bảo rằng:
- Bạn đã chạy
pip install "scrapling[fetchers]" - Bạn đã thực thi
scrapling installđể tải xuống các phụ thuộc của trình duyệt - Bạn có Scrapeless API Key hợp lệ
- Bạn đang sử dụng Python 3.10+
Thu thập dữ liệu Amazon với Scrapling + Scrapeless
Dưới đây là một ví dụ hoàn chỉnh về việc thu thập dữ liệu chi tiết sản phẩm Amazon.
Tập lệnh tự động kết nối với Scrapeless Cloud Browser, tải trang đích, vượt qua các kiểm tra chống bot và trích xuất thông tin sản phẩm chính — chẳng hạn như tiêu đề, giá cả, tình trạng kho hàng, xếp hạng, số lượng đánh giá, tính năng, hình ảnh, ASIN, nhà cung cấp và danh mục.
# amazon_scraper_response_only.py
from urllib.parse import urlencode
import json
import time
import re
from scrapling.fetchers import DynamicSession
# ---------------- CONFIG ----------------
CONFIG = {
"token": "YOUR_SCRAPELESS_API_KEY",
"sessionName": "Data Scraping",
"sessionTTL": "900",
"proxyCountry": "ANY",
"sessionRecording": "true",
}
DISABLE_RESOURCES = True # False -> load JS/resources (more stable for JS-heavy sites)
WAIT_FOR_SELECTOR_TIMEOUT = 60
MAX_RETRIES = 3
TARGET_URL = "https://www.amazon.com/ESR-Compatible-Military-Grade-Protection-Scratch-Resistant/dp/B0CC1F4V7Q"
WS_ENDPOINT = f"wss://browser.scrapeless.com/api/v2/browser?{urlencode(CONFIG)}"
# ---------------- HELPERS (use response ONLY) ----------------
def retry(func, retries=2, wait=2):
for i in range(retries + 1):
try:
return func()
except Exception as e:
print(f"[retry] Attempt {i+1} failed: {e}")
if i == retries:
raise
time.sleep(wait * (i + 1))
def _resp_css_first_text(resp, selector):
"""Try response.css_first('selector::text') or resp.query_selector_text(selector) - return str or None."""
try:
if hasattr(resp, "css_first"):
# prefer unified ::text pseudo API
val = resp.css_first(f"{selector}::text")
if val:
return val.strip()
except Exception:
pass
try:
if hasattr(resp, "query_selector_text"):
val = resp.query_selector_text(selector)
if val:
return val.strip()
except Exception:
pass
return None
def _resp_css_texts(resp, selector):
"""Return list of text values for selector using response.css('selector::text') or query_selector_all_text."""
out = []
try:
if hasattr(resp, "css"):
vals = resp.css(f"{selector}::text") or []
for v in vals:
if isinstance(v, str) and v.strip():
out.append(v.strip())
if out:
return out
except Exception:
pass
try:
if hasattr(resp, "query_selector_all_text"):
vals = resp.query_selector_all_text(selector) or []
for v in vals:
if v and v.strip():
out.append(v.strip())
if out:
return out
except Exception:
pass
# some fetchers provide query_selector_all and elements with .text() method
try:
if hasattr(resp, "query_selector_all"):
els = resp.query_selector_all(selector) or []
for el in els:
try:
if hasattr(el, "text") and callable(el.text):
t = el.text()
if t and t.strip():
out.append(t.strip())
continue
except Exception:
pass
try:
if hasattr(el, "get_text"):
t = el.get_text(strip=True)
if t:
out.append(t)
continue
except Exception:
pass
except Exception:
pass
return out
def _resp_css_first_attr(resp, selector, attr):
"""Try to get attribute via response css pseudo ::attr(...) or query selector element attributes."""
try:
if hasattr(resp, "css_first"):
val = resp.css_first(f"{selector}::attr({attr})")
if val:
return val.strip()
except Exception:
pass
try:
# try element and get_attribute / get
if hasattr(resp, "query_selector"):
el = resp.query_selector(selector)
if el:
if hasattr(el, "get_attribute"):
try:
v = el.get_attribute(attr)
if v:
return v
except Exception:
pass
try:
v = el.get(attr) if hasattr(el, "get") else None
if v:
return v
except Exception:
pass
try:
attrs = getattr(el, "attrs", None)
if isinstance(attrs, dict) and attr in attrs:
return attrs.get(attr)
except Exception:
pass
except Exception:
pass
return None
def detect_bot_via_resp(resp):
"""Detect typical bot/captcha signals using response text selectors only."""
checks = [
# body text
("body",),
# some common challenge indicators
("#challenge-form",),
("#captcha",),
("text:contains('are you a human')",),
]
# First try a broad body text
try:
body_text = _resp_css_first_text(resp, "body")
if body_text:
txt = body_text.lower()
for k in ("captcha", "are you a human", "verify you are human", "access to this page has been denied", "bot detection", "please enable javascript", "checking your browser"):
if k in txt:
return True
except Exception:
pass
# Try specific selectors
suspects = [
"#captcha", "#cf-hcaptcha-container", "#challenge-form", "text:contains('are you a human')"
]
for s in suspects:
try:
if _resp_css_first_text(resp, s):
return True
except Exception:
pass
return False
def parse_price_from_text(price_raw):
if not price_raw:
return None, None
m = re.search(r"([^\d.,\s]+)?\s*([\d,]+\.\d{1,2}|[\d,]+)", price_raw)
if m:
currency = m.group(1).strip() if m.group(1) else None
num = m.group(2).replace(",", "")
try:
price = float(num)
except Exception:
price = None
return currency, price
return None, None
def parse_int_from_text(text):
if not text:
return None
digits = "".join(filter(str.isdigit, text))
try:
return int(digits) if digits else None
except:
return None
# ---------------- MAIN (use response only) ----------------
def scrape_amazon_using_response_only(url):
with DynamicSession(cdp_url=WS_ENDPOINT, disable_resources=DISABLE_RESOURCES) as s:
# fetch with retry
resp = retry(lambda: s.fetch(url, network_idle=True, timeout=120000), retries=MAX_RETRIES - 1)
if detect_bot_via_resp(resp):
print("[warn] Bot/CAPTCHA detected via response selectors.")
try:
resp.screenshot(path="captcha_detected.png")
except Exception:
pass
# retry once
time.sleep(2)
resp = retry(lambda: s.fetch(url, network_idle=True, timeout=120000), retries=1)
# Wait for productTitle (polling using resp selectors only)
title = _resp_css_first_text(resp, "#productTitle") or _resp_css_first_text(resp, "#title")
waited = 0
while not title and waited < WAIT_FOR_SELECTOR_TIMEOUT:
print("[info] Waiting for #productTitle to appear (response selector)...")
time.sleep(3)
waited += 3
resp = s.fetch(url, network_idle=True, timeout=120000)
title = _resp_css_first_text(resp, "#productTitle") or _resp_css_first_text(resp, "#title")
title = title.strip() if title else None
# Extract fields using response-only helpers
def get_text(selectors, multiple=False):
if multiple:
out = []
for sel in selectors:
out.extend(_resp_css_texts(resp, sel) or [])
return out
for sel in selectors:
v = _resp_css_first_text(resp, sel)
if v:
return v
return None
price_raw = get_text([
"#priceblock_ourprice",
"#priceblock_dealprice",
"#priceblock_saleprice",
"#price_inside_buybox",
".a-price .a-offscreen"
])
rating_text = get_text(["span.a-icon-alt", "#acrPopover"])
review_count_text = get_text(["#acrCustomerReviewText", "[data-hook='total-review-count']"])
availability = get_text([
"#availability .a-color-state",
"#availability .a-color-success",
"#outOfStock",
"#availability"
])
features = get_text(["#feature-bullets ul li"], multiple=True) or []
description = get_text([
"#productDescription",
"#bookDescription_feature_div .a-expander-content",
"#productOverview_feature_div"
])
# images (use attribute extraction via response)
images = []
seen = set()
main_src = _resp_css_first_attr(resp, "#imgTagWrapperId img", "data-old-hires") \
or _resp_css_first_attr(resp, "#landingImage", "src") \
or _resp_css_first_attr(resp, "#imgTagWrapperId img", "src")
if main_src and main_src not in seen:
images.append(main_src); seen.add(main_src)
dyn = _resp_css_first_attr(resp, "#imgTagWrapperId img", "data-a-dynamic-image") \
or _resp_css_first_attr(resp, "#landingImage", "data-a-dynamic-image")
if dyn:
try:
obj = json.loads(dyn)
for k in obj.keys():
if k not in seen:
images.append(k); seen.add(k)
except Exception:
pass
thumbs = _resp_css_texts(resp, "#altImages img::attr(src)") or _resp_css_texts(resp, ".imageThumbnail img::attr(src)") or []
for src in thumbs:
if not src:
continue
src_clean = re.sub(r"\._[A-Z0-9,]+_\.", ".", src)
if src_clean not in seen:
images.append(src_clean); seen.add(src_clean)
# ASIN (attribute)
asin = _resp_css_first_attr(resp, "input#ASIN", "value")
if asin:
asin = asin.strip()
else:
detail_texts = _resp_css_texts(resp, "#detailBullets_feature_div li") or []
combined = " ".join([t for t in detail_texts if t])
m = re.search(r"ASIN[:\s]*([A-Z0-9-]+)", combined, re.I)
if m:
asin = m.group(1).strip()
merchant = _resp_css_first_text(resp, "#sellerProfileTriggerId") \
or _resp_css_first_text(resp, "#merchant-info") \
or _resp_css_first_text(resp, "#bylineInfo")
categories = _resp_css_texts(resp, "#wayfinding-breadcrumbs_container ul li a") or _resp_css_texts(resp, "#wayfinding-breadcrumbs_feature_div ul li a") or []
categories = [c.strip() for c in categories if c and c.strip()]
currency, price = parse_price_from_text(price_raw)
rating_val = None
if rating_text:
try:
rating_val = float(rating_text.split()[0].replace(",", ""))
except Exception:
rating_val = None
review_count = parse_int_from_text(review_count_text)
data = {
"title": title,
"price_raw": price_raw,
"price": price,
"currency": currency,
"rating": rating_val,
"review_count": review_count,
"availability": availability,
"features": features,
"description": description,
"images": images,
"asin": asin,
"merchant": merchant,
"categories": categories,
"url": url,
"scrapedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
return data
# ---------------- RUN ----------------
if __name__ == "__main__":
try:
result = scrape_amazon_using_response_only(TARGET_URL)
print(json.dumps(result, indent=2, ensure_ascii=False))
with open("scrapeless-amazon-product.json", "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
except Exception as e:
print("[error] scraping failed:", e)Ví dụ đầu ra:
{
"title": "ESR for iPhone 15 Pro Max Case, Compatible with MagSafe, Military-Grade Protection, Yellowing Resistant, Scratch-Resistant Back, Magnetic Phone Case for iPhone 15 Pro Max, Classic Series, Clear",
"price_raw": "$12.99",
"price": 12.99,
"currency": "$",
"rating": 4.6,
"review_count": 133714,
"availability": "In Stock",
"features": [
"Compatibility: only for iPhone 15 Pro Max; full functionality maintained via precise speaker and port cutouts and easy-press buttons",
"Stronger Magnetic Lock: powerful built-in magnets with 1,500 g of holding force enable faster, easier place-and-go wireless charging and a secure lock on any MagSafe accessory",
"Military-Grade Drop Protection: rigorously tested to ensure total protection on all sides, with specially designed Air Guard corners that absorb shock so your phone doesn\u2019t have to",
"Raised-Edge Protection: raised screen edges and Camera Guard lens frame provide enhanced scratch protection where it really counts",
"Stay Original: scratch-resistant, crystal-clear acrylic back lets you show off your iPhone 15 Pro Max\u2019s true style in stunning clarity that lasts",
"Complete Customer Support: detailed setup videos and FAQs, comprehensive 12-month protection plan, lifetime support, and personalized help."
],
"description": "BrandESRCompatible Phone ModelsiPhone 15 Pro MaxColorA-ClearCompatible DevicesiPhone 15 Pro MaxMaterialAcrylic",
"images": [
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SL1500_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX342_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX679_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX522_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX385_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX466_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX425_.jpg",
"https://m.media-amazon.com/images/I/71-ishbNM+L._AC_SX569_.jpg",
"https://m.media-amazon.com/images/I/41Ajq9jnx9L._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/51RkuGXBMVL._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/516RCbMo5tL._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/51DdOFdiQQL._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/514qvXYcYOL._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/518CS81EFXL._AC_SR38,50_.jpg",
"https://m.media-amazon.com/images/I/413EWAtny9L.SX38_SY50_CR,0,0,38,50_BG85,85,85_BR-120_PKdp-play-icon-overlay__.jpg",
"https://images-na.ssl-images-amazon.com/images/G/01/x-locale/common/transparent-pixel._V192234675_.gif"
],
"asin": "B0CC1F4V7Q",
"merchant": "Minghutech-US",
"categories": [
"Cell Phones & Accessories",
"Cases, Holsters & Sleeves",
"Basic Cases"
],
"url": "https://www.amazon.com/ESR-Compatible-Military-Grade-Protection-Scratch-Resistant/dp/B0CC1F4V7Q",
"scrapedAt": "2025-10-30T10:20:16Z"
}Ví dụ này minh họa cách DynamicSession và Scrapeless có thể hoạt động cùng nhau để tạo ra một môi trường phiên dài ổn định, có thể tái sử dụng.
Trong cùng một phiên, bạn có thể yêu cầu nhiều trang mà không cần khởi động lại trình duyệt, duy trì trạng thái đăng nhập, cookie và bộ nhớ cục bộ, đồng thời đạt được cô lập hồ sơ và duy trì phiên.