Scrapy là một trong những framework Python mạnh mẽ nhất cho việc web scraping, nhưng nếu không có cài đặt proxy đúng cách, các trình phân tích của bạn sẽ bị chặn chỉ sau vài phút hoạt động. Trong hướng dẫn này, tôi sẽ chỉ cho bạn tất cả các cách tích hợp proxy vào Scrapy: từ cài đặt đơn giản nhất đến các phương pháp nâng cao xoay vòng địa chỉ IP với xử lý lỗi tự động.
Tài liệu này dựa trên kinh nghiệm thực tế trong việc phân tích các trang thương mại điện tử lớn và các trang web bảo mật. Bạn sẽ nhận được các ví dụ mã sẵn sàng để sử dụng ngay trong các dự án của mình.
Tại sao Scrapy không có proxy lại bị chặn
Các trang web hiện đại sử dụng nhiều lớp bảo vệ chống lại việc phân tích. Ngay cả khi bạn đã cài đặt User-Agent và thời gian chờ giữa các yêu cầu, địa chỉ IP của bạn sẽ bị phát hiện tự động hóa qua một số dấu hiệu:
- Tần suất yêu cầu: một IP thực hiện 100+ yêu cầu mỗi phút — dấu hiệu rõ ràng của bot
- Mô hình hành vi: duyệt các trang liên tục mà không có các chuyển tiếp ngẫu nhiên
- Thiếu JavaScript: Scrapy không thực thi JS, điều này dễ dàng bị phát hiện
- Định vị địa lý: truy cập từ trung tâm dữ liệu thay vì mạng gia đình
Kết quả — bị cấm theo IP trong vài giờ hoặc vài ngày. Các trang thương mại điện tử (Amazon, Wildberries, Ozon), mạng xã hội và các trang web với Cloudflare sử dụng bảo vệ rất nghiêm ngặt. Proxy giải quyết vấn đề này bằng cách phân phối yêu cầu giữa nhiều địa chỉ IP.
Quan trọng: Ngay cả với proxy, bạn cũng cần tuân thủ giới hạn tần suất. Tốc độ khuyến nghị: 1-3 yêu cầu mỗi giây cho một IP. Để phân tích tốc độ cao, hãy sử dụng một nhóm hơn 50 proxy với xoay vòng.
Cài đặt cơ bản proxy trong Scrapy
Cách đơn giản nhất là chỉ định proxy trực tiếp trong cài đặt của con nhện. Phương pháp này phù hợp cho việc thử nghiệm hoặc phân tích một lượng dữ liệu nhỏ với một máy chủ proxy.
Phương pháp 1: Qua meta trong Request
import scrapy
class MySpider(scrapy.Spider):
name = 'example'
start_urls = ['https://example.com']
def start_requests(self):
proxy = 'http://username:password@proxy.example.com:8080'
for url in self.start_urls:
yield scrapy.Request(
url=url,
callback=self.parse,
meta={'proxy': proxy}
)
def parse(self, response):
# Logic phân tích của bạn
self.log(f'Scraped {response.url} qua {response.meta["proxy"]}')Định dạng proxy phụ thuộc vào giao thức và phương pháp xác thực:
http://proxy.example.com:8080— không có xác thựchttp://user:pass@proxy.example.com:8080— có tên đăng nhập/mật khẩusocks5://user:pass@proxy.example.com:1080— proxy SOCKS5
Phương pháp 2: Cài đặt toàn cầu trong settings.py
# settings.py
# HTTP proxy cho tất cả các yêu cầu
HTTPPROXY_ENABLED = True
HTTPPROXY_AUTH_ENCODING = 'utf-8'
# Cài đặt qua biến môi trường
HTTP_PROXY = 'http://username:password@proxy.example.com:8080'
HTTPS_PROXY = 'http://username:password@proxy.example.com:8080'Phương pháp này tiện lợi cho việc thử nghiệm nhanh, nhưng không phù hợp cho sản xuất: không có xoay vòng IP, nếu proxy bị hỏng, toàn bộ trình phân tích sẽ dừng lại, không thể sử dụng các proxy khác nhau cho các trang web khác nhau.
Tạo Middleware Proxy tùy chỉnh
Để phân tích sản xuất, cần một middleware riêng, sẽ quản lý nhóm proxy, xử lý lỗi và tự động xoay vòng IP. Đây là một triển khai cơ bản:
# middlewares.py
import random
from scrapy import signals
from scrapy.exceptions import NotConfigured
class RandomProxyMiddleware:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
@classmethod
def from_crawler(cls, crawler):
# Tải danh sách proxy từ cài đặt
proxy_list = crawler.settings.getlist('PROXY_LIST')
if not proxy_list:
raise NotConfigured('PROXY_LIST chưa được cấu hình')
return cls(proxy_list)
def process_request(self, request, spider):
# Chọn một proxy ngẫu nhiên từ nhóm
proxy = random.choice(self.proxy_list)
request.meta['proxy'] = proxy
spider.logger.info(f'Sử dụng proxy: {proxy}')
def process_exception(self, request, exception, spider):
# Khi có lỗi, thử một proxy khác
proxy = random.choice(self.proxy_list)
request.meta['proxy'] = proxy
spider.logger.warning(
f'Lỗi proxy, chuyển sang: {proxy}'
)
return request
Bây giờ hãy cấu hình việc sử dụng middleware trong settings.py:
# settings.py
# Danh sách proxy (có thể tải từ tệp hoặc API)
PROXY_LIST = [
'http://user1:pass1@proxy1.example.com:8080',
'http://user2:pass2@proxy2.example.com:8080',
'http://user3:pass3@proxy3.example.com:8080',
# ... thêm 50+ proxy để xoay vòng hiệu quả
]
# Kết nối middleware
DOWNLOADER_MIDDLEWARES = {
'myproject.middlewares.RandomProxyMiddleware': 350,
'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 400,
}
# Thử lại khi có lỗi
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]Xoay vòng proxy: ba phương pháp hiệu quả
Chọn proxy ngẫu nhiên (như trong ví dụ trên) là phương pháp đơn giản nhất, nhưng không hiệu quả nhất. Hãy xem xét ba chiến lược xoay vòng cho các kịch bản khác nhau.
Phương pháp 1: Round-robin (xoay vòng tuần tự)
Proxy được chọn theo vòng tròn. Phù hợp cho việc phân phối tải đều:
class RoundRobinProxyMiddleware:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.current_index = 0
@classmethod
def from_crawler(cls, crawler):
proxy_list = crawler.settings.getlist('PROXY_LIST')
return cls(proxy_list)
def process_request(self, request, spider):
# Lấy proxy tiếp theo theo vòng tròn
proxy = self.proxy_list[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxy_list)
request.meta['proxy'] = proxyPhương pháp 2: Xoay vòng thông minh với danh sách đen
Theo dõi các proxy gặp vấn đề và tạm thời loại bỏ chúng khỏi xoay vòng:
import time
from collections import defaultdict
class SmartProxyMiddleware:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.proxy_errors = defaultdict(int)
self.blacklist = set()
self.blacklist_timeout = 300 # 5 phút
self.blacklist_time = {}
@classmethod
def from_crawler(cls, crawler):
proxy_list = crawler.settings.getlist('PROXY_LIST')
return cls(proxy_list)
def get_working_proxies(self):
# Loại bỏ các proxy trong danh sách đen, thời gian đã hết
current_time = time.time()
expired = [
proxy for proxy, ban_time in self.blacklist_time.items()
if current_time - ban_time > self.blacklist_timeout
]
for proxy in expired:
self.blacklist.discard(proxy)
self.proxy_errors[proxy] = 0
# Trả về các proxy hoạt động
return [p for p in self.proxy_list if p not in self.blacklist]
def process_request(self, request, spider):
working_proxies = self.get_working_proxies()
if not working_proxies:
spider.logger.error('Tất cả các proxy đều bị đưa vào danh sách đen!')
return
proxy = random.choice(working_proxies)
request.meta['proxy'] = proxy
def process_response(self, request, response, spider):
# Nếu bị chặn — thêm vào danh sách đen
if response.status in [403, 429, 503]:
proxy = request.meta.get('proxy')
self.proxy_errors[proxy] += 1
if self.proxy_errors[proxy] >= 3:
self.blacklist.add(proxy)
self.blacklist_time[proxy] = time.time()
spider.logger.warning(
f'Proxy {proxy} đã bị đưa vào danh sách đen trong {self.blacklist_timeout}s'
)
return responsePhương pháp 3: Xoay vòng qua API của nhà cung cấp
Nhiều nhà cung cấp proxy (bao gồm proxy cư trú) cung cấp endpoint xoay vòng — một URL, tự động thay đổi IP mỗi khi có yêu cầu:
# settings.py
# Một endpoint duy nhất với xoay vòng tự động
ROTATING_PROXY = 'http://username:password@rotating.proxy.com:8080'
# Middleware đơn giản
class RotatingProxyMiddleware:
def __init__(self, proxy):
self.proxy = proxy
@classmethod
def from_crawler(cls, crawler):
proxy = crawler.settings.get('ROTATING_PROXY')
return cls(proxy)
def process_request(self, request, spider):
# Một URL, nhưng mỗi yêu cầu đi với một IP mới
request.meta['proxy'] = self.proxyĐây là phương pháp tiện lợi nhất cho sản xuất: không cần quản lý nhóm proxy, nhà cung cấp tự theo dõi chất lượng IP và thay thế các proxy gặp vấn đề. Đặc biệt hiệu quả với proxy cư trú, nơi nhóm IP có thể đạt tới hàng triệu địa chỉ.
Xác thực: tên đăng nhập/mật khẩu so với danh sách trắng IP
Các nhà cung cấp proxy cung cấp hai phương pháp xác thực. Sự lựa chọn ảnh hưởng đến tốc độ kết nối và sự tiện lợi trong cài đặt.
Xác thực User:Pass
Tên đăng nhập và mật khẩu được truyền trong URL của proxy. Scrapy tự động chuyển đổi chúng thành tiêu đề HTTP Proxy-Authorization:
proxy = 'http://username:password@proxy.example.com:8080'
request.meta['proxy'] = proxy
# Scrapy sẽ tự động thêm tiêu đề:
# Proxy-Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=
Ưu điểm: hoạt động từ bất kỳ IP nào, dễ dàng thay đổi proxy trong mã.
Nhược điểm: một chút overhead cho mỗi yêu cầu (~50-100ms), thông tin xác thực ở dạng rõ trong mã.
Xác thực danh sách trắng IP
Bạn thêm IP của máy chủ của bạn vào danh sách trắng của nhà cung cấp, không cần xác thực:
proxy = 'http://proxy.example.com:8080' # không có tên đăng nhập/mật khẩu
request.meta['proxy'] = proxy
Ưu điểm: nhanh hơn 50-100ms, an toàn hơn (không có thông tin xác thực trong mã).
Nhược điểm: chỉ hoạt động từ các IP nhất định, cần cập nhật danh sách trắng khi thay đổi máy chủ.
Khuyến nghị cho sản xuất:
Sử dụng danh sách trắng IP cho việc phân tích từ các máy chủ chuyên dụng (AWS, Google Cloud, Hetzner). Đối với phát triển và thử nghiệm từ máy tính cục bộ — sử dụng xác thực user:pass.
Xử lý lỗi và tự động thay đổi IP
Ngay cả với các proxy chất lượng, sẽ có lỗi: timeout, từ chối kết nối, chặn. Xử lý lỗi đúng cách là rất quan trọng cho sự ổn định của trình phân tích.
Xử lý mã trạng thái HTTP
class ProxyMiddleware:
def process_response(self, request, response, spider):
# Các mã, khi cần thay đổi proxy và thử lại
ban_codes = [403, 407, 429, 503]
if response.status in ban_codes:
proxy = request.meta.get('proxy')
spider.logger.warning(
f'Nhận {response.status} từ {proxy}, đang thử lại...'
)
# Đánh dấu để thử lại với proxy mới
request.meta['dont_retry'] = False
request.meta['proxy'] = self.get_new_proxy()
return request
return responseXử lý ngoại lệ mạng
from twisted.internet.error import TimeoutError, ConnectionRefusedError
from scrapy.exceptions import IgnoreRequest
class ProxyMiddleware:
def process_exception(self, request, exception, spider):
# Lỗi kết nối đến proxy
proxy_errors = (
TimeoutError,
ConnectionRefusedError,
ConnectionLost,
)
if isinstance(exception, proxy_errors):
proxy = request.meta.get('proxy')
spider.logger.error(
f'Kết nối proxy {proxy} thất bại: {exception}'
)
# Thay đổi proxy và thử lại
request.meta['proxy'] = self.get_new_proxy()
return request
# Đối với các lỗi khác, sử dụng xử lý tiêu chuẩn
return NonePhát hiện chặn qua nội dung
Một số trang web trả về HTTP 200, nhưng hiển thị captcha hoặc trang chặn:
class ProxyMiddleware:
def process_response(self, request, response, spider):
# Dấu hiệu chặn trong nội dung
ban_indicators = [
'captcha',
'truy cập bị từ chối',
'bị chặn',
'lưu lượng không bình thường',
'kiểm tra robot',
]
body_text = response.text.lower()
if any(indicator in body_text for indicator in ban_indicators):
spider.logger.warning(
f'Trái phép trang bị phát hiện từ {request.meta.get("proxy")}'
)
# Thay đổi proxy và thử lại
request.meta['proxy'] = self.get_new_proxy()
return request
return responseLoại proxy nào nên chọn cho Scrapy
Việc chọn loại proxy phụ thuộc vào trang web mục tiêu, ngân sách và tốc độ phân tích yêu cầu. Dưới đây là so sánh các tùy chọn chính:
| Loại proxy | Tốc độ | Chi phí | Khi nào sử dụng |
|---|---|---|---|
| Proxy trung tâm dữ liệu | Cao (50-200ms) | Thấp ($1-3/IP) | Các trang đơn giản không có bảo vệ, API, công cụ nội bộ |
| Proxy cư trú | Trung bình (300-800ms) | Trung bình ($5-15/GB) | Thương mại điện tử, mạng xã hội, các trang với Cloudflare, nhắm mục tiêu địa lý |
| Proxy di động | Thấp (500-1500ms) | Cao ($50-150/IP) | Ứng dụng di động, Instagram, TikTok, bảo vệ tối đa |
Khuyến nghị về lựa chọn
Đối với việc phân tích các trang thương mại điện tử (Amazon, Wildberries, Ozon, AliExpress) — chỉ sử dụng proxy cư trú. Những trang này thường xuyên chặn các trung tâm dữ liệu. Cần có xoay vòng và nhắm mục tiêu địa lý (ví dụ, IP Nga cho Wildberries).
Đối với việc phân tích các trang tin tức, blog, diễn đàn — proxy trung tâm dữ liệu là đủ. Bảo vệ tối thiểu, tốc độ và chi phí lưu lượng quan trọng.
Đối với việc phân tích các trang với Cloudflare — proxy cư trú là bắt buộc. Các trung tâm dữ liệu bị Cloudflare phát hiện gần như ngay lập tức. Thêm thư viện cloudscraper vào Scrapy để vượt qua các thử thách JS.
Đối với việc phân tích Google Search, công cụ SEO — proxy cư trú với nhắm mục tiêu địa lý. Google hiển thị các kết quả khác nhau cho các quốc gia và thành phố khác nhau.
Lời khuyên: Bắt đầu với một nhóm 10 proxy cư trú để thử nghiệm. Nếu bạn gặp phải các chặn — tăng nhóm lên 50-100 IP. Đối với việc phân tích tốc độ cao (1000+ yêu cầu/phút) hãy sử dụng endpoint xoay vòng với nhóm 10,000+ IP.
Kỹ thuật nâng cao: phiên và IP dính
Khi phân tích một số trang web, cần giữ một IP trong suốt phiên (xác thực, giỏ hàng, các biểu mẫu nhiều bước). Đây là cách thực hiện các phiên dính trong Scrapy.
IP dính cho một miền
from urllib.parse import urlparse
class StickyProxyMiddleware:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
# Từ điển: miền -> proxy
self.domain_proxy_map = {}
@classmethod
def from_crawler(cls, crawler):
proxy_list = crawler.settings.getlist('PROXY_LIST')
return cls(proxy_list)
def process_request(self, request, spider):
# Trích xuất miền từ URL
domain = urlparse(request.url).netloc
# Nếu đã có proxy cho miền này — sử dụng nó
if domain in self.domain_proxy_map:
proxy = self.domain_proxy_map[domain]
else:
# Nếu không, chọn mới và ghi nhớ
proxy = random.choice(self.proxy_list)
self.domain_proxy_map[domain] = proxy
spider.logger.info(f'Gán {proxy} cho {domain}')
request.meta['proxy'] = proxyIP dính với thời gian phiên
Một biến thể nâng cao hơn: proxy được gán cho miền trong một khoảng thời gian nhất định (ví dụ, 10 phút), sau đó sẽ thay đổi:
import time
from urllib.parse import urlparse
class SessionProxyMiddleware:
def __init__(self, proxy_list, session_timeout=600):
self.proxy_list = proxy_list
self.session_timeout = session_timeout # 10 phút
# Từ điển: miền -> (proxy, thời gian tạo)
self.sessions = {}
@classmethod
def from_crawler(cls, crawler):
proxy_list = crawler.settings.getlist('PROXY_LIST')
timeout = crawler.settings.getint('PROXY_SESSION_TIMEOUT', 600)
return cls(proxy_list, timeout)
def get_proxy_for_domain(self, domain):
current_time = time.time()
# Kiểm tra xem có phiên hoạt động không
if domain in self.sessions:
proxy, created_at = self.sessions[domain]
# Nếu phiên chưa hết hạn — sử dụng cùng một proxy
if current_time - created_at < self.session_timeout:
return proxy
# Tạo một phiên mới với proxy mới
new_proxy = random.choice(self.proxy_list)
self.sessions[domain] = (new_proxy, current_time)
return new_proxy
def process_request(self, request, spider):
domain = urlparse(request.url).netloc
proxy = self.get_proxy_for_domain(domain)
request.meta['proxy'] = proxyTích hợp với Cookie Middleware
Để có các phiên đầy đủ, cần đồng bộ hóa proxy và cookies. Scrapy lưu trữ cookies riêng cho mỗi miền, nhưng khi thay đổi proxy cần xóa cookies:
# settings.py
# Bật cookie middleware
COOKIES_ENABLED = True
COOKIES_DEBUG = False
# Middleware để đồng bộ hóa proxy và cookies
class ProxyCookieMiddleware:
def process_request(self, request, spider):
# Lấy proxy hiện tại
current_proxy = request.meta.get('proxy')
# Nếu proxy đã thay đổi — xóa cookies
previous_proxy = request.meta.get('previous_proxy')
if previous_proxy and previous_proxy != current_proxy:
# Xóa cookies cho miền này
jar = spider.crawler.engine.downloader.middleware.middlewares[0].jars
domain = urlparse(request.url).netloc
if domain in jar:
jar[domain].clear()
spider.logger.info(f'Đã xóa cookies cho {domain}')
request.meta['previous_proxy'] = current_proxyKết luận
Cài đặt proxy đúng cách trong Scrapy là nền tảng cho việc phân tích ổn định mà không bị chặn. Chúng ta đã xem xét tất cả các khía cạnh chính: từ tích hợp cơ bản đến các kỹ thuật nâng cao về xoay vòng và quản lý phiên.
Những điểm chính:
- Đối với sản xuất, hãy sử dụng middleware tùy chỉnh với xoay vòng thông minh và danh sách đen các IP gặp vấn đề
- Xử lý tất cả các loại lỗi: mã trạng thái HTTP, ngoại lệ mạng, chặn theo nội dung
- Chọn loại proxy phù hợp với nhiệm vụ: trung tâm dữ liệu cho các trang đơn giản, cư trú cho các trang bảo mật
- Đối với các trang yêu cầu xác thực, hãy sử dụng các phiên dính với proxy gán cho miền
- Bắt đầu với nhóm 10-50 proxy, mở rộng khi tải tăng lên
Nếu bạn dự định phân tích các trang web bảo mật (thương mại điện tử, mạng xã hội, các trang với Cloudflare), tôi khuyên bạn nên sử dụng proxy cư trú — chúng cung cấp tính ẩn danh tối đa và rủi ro bị chặn tối thiểu. Đối với phân tích tốc độ cao, hãy chọn các nhà cung cấp có endpoint xoay vòng và nhóm từ 10,000 địa chỉ IP.
Tất cả các ví dụ mã trong bài viết này đã được kiểm tra trên Scrapy 2.x và sẵn sàng để sử dụng trong sản xuất. Hãy điều chỉnh chúng theo nhu cầu của bạn và mở rộng khi dự án phát triển.