爬虫
爬虫基础
最基本的网页爬虫,只有两个关键点:
- 模拟HTTP请求
- 解析HTML页面
其中解析HTML页面,由于非常的case-by-case(页面结构/需求都很灵活),所以我们一般使用灵活的语言,如Python进行开发。下面介绍如何用Python完成一个基本的爬虫。
Request - 模拟HTTP请求
request是 Python 久负盛名的 HTTP 库。接口设计非常的人性化。下面是一个Demo,(i) 通过headers简单模拟浏览器请求,(ii) 使用代理,(iii) 简单的流量控制、重试和错误处理。
import requests
from requests.exceptions import RequestException
import logging
from time import sleep
import random
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 多个 User-Agent 选项
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
]
# 代理设置
PROXIES = {
'http': 'http://127.0.0.1:7890',
'https': 'http://127.0.0.1:7890',
}
def get_random_user_agent():
return random.choice(USER_AGENTS)
def proxy_get(url, max_retries=5, base_delay=1, proxies=PROXIES):
headers = {'User-Agent': get_random_user_agent()}
for i in range(max_retries):
try:
response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
response.raise_for_status() # 这将抛出 HTTPError,如果状态码不是 200
return response.content
except RequestException as e:
logger.error(f"Attempt {i+1}/{max_retries} failed for {url}. Error: {str(e)}")
if i < max_retries - 1: # 不是最后一次尝试
sleep_time = base_delay * (2 ** i) + random.uniform(0, 1) # 指数退避 + 随机化
logger.info(f"Retrying in {sleep_time:.2f} seconds...")
sleep(sleep_time)
else:
logger.error(f"All retries failed for {url}.")
return None
# 使用示例
if __name__ == "__main__":
test_url = "https://example.com"
content = proxy_get(test_url)
if content:
logger.info(f"Successfully retrieved content from {test_url}")
else:
logger.error(f"Failed to retrieve content from {test_url}")
Lxml - 解析HTML页面
我一般使用 xpath 梭哈所有的HTML解析工作。 xpath 的设计思路就是把HTML歇息成一棵树,然后从根开始逐层筛选节点。下面是一个例子
from lxml import html
tree = html.fromstring(content)
rows = tree.xpath('//*[@id="ContentPlaceHolder1_tbodyTxnTable"]//tr')
'//*[@id="ContentPlaceHolder1_tbodyTxnTable"]'
的意思是:(根节点)的子孙节点中,任意id
属性等于ContentPlaceHolder1_tbodyTxnTable
的节点。...//tr
:(...)的子孙tr
节点。
xpath 的语法并不复杂,看一下Xpath Cheatsheet就知道个七七八八,够用了。一般用的最多的是 Descendant selectors 和 Attribute selectors 。
Advanced
考虑到现实对效率的追求,以及网页的反爬虫手段,实际的爬虫工程还需要更多的内容
- 多线程并发爬虫(由于是IO bounding,所以使用异步可能会有更高的效率)
- 拟人(反反爬虫):网页终归是人来访问的,所以总是可以爬的
- JavaScript模拟
- 验证码
- 身份验证:例如Cookie
异步并发
httpx 可以说是 requests 的增强复杂版本,它提供了对包括异步在内的很多特性支持,并且借鉴了 requests 的接口设计。给一个demo,基于上面 request demo的修改版。
import httpx
import random
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# User-Agent 列表
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
]
# 代理设置
PROXIES = {
'http://': 'http://127.0.0.1:7890',
'https://': 'http://127.0.0.1:7890',
}
def get_random_user_agent():
return random.choice(USER_AGENTS)
def fetch_url(url, max_retries=3):
headers = {'User-Agent': get_random_user_agent()}
for attempt in range(max_retries):
try:
with httpx.Client(proxies=PROXIES, timeout=10) as client:
response = client.get(url, headers=headers)
response.raise_for_status()
return response.content
except httpx.HTTPError as e:
logger.error(f"Attempt {attempt + 1}/{max_retries} failed for {url}. Error: {str(e)}")
if attempt == max_retries - 1:
logger.error(f"All retries failed for {url}.")
return None
def fetch_all_urls(urls, max_workers=10):
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
results[url] = data
if data:
logger.info(f"Successfully retrieved content from {url}")
else:
logger.error(f"Failed to retrieve content from {url}")
except Exception as exc:
logger.error(f'{url} generated an exception: {exc}')
return results
# 使用示例
if __name__ == "__main__":
test_urls = [
"https://example.com",
"https://example.org",
"https://example.net"
]
start_time = time.time()
results = fetch_all_urls(test_urls)
end_time = time.time()
logger.info(f"Total time taken: {end_time - start_time:.2f} seconds")
logger.info(f"Number of successful requests: {sum(1 for content in results.values() if content is not None)}")
JavaScript 模拟
分两种情况:
- 数据是通过JavaScript在前端渲染出来的:这个没有什么好办法,一般使用无头浏览器对网页进行请求,再作进一步分析。
- 数据是通过JavaScript,通过AJAX获取的:一般情况下,可以找到后端对应的API,API返回数据一般是结构性的,反而省略了解析HTML页面的步骤。寻找API有两种办法,一种是抓包,另一种是看JavaScript源码。
Selenium 是模拟浏览器的常用解决方案。不过要模拟浏览器的话,效率就不会太高了。一般是用来进行测试而不是爬虫的。
验证码
验证码本来就是要验证你是不是人的,自然是没那么好绕过的。最管用的方法是买第三方服务,简单的方法是自己点验证码(。
不过有的时候,服务器只有在检测到异常流量才会要求输入验证码。这时可以使用流量控制、IP池、云服务等方法,防止单个IP快速发送请求,被服务器发现异常。
身份验证
Cookie: 临时应付一下的话,直接把浏览器访问的Cookie字段拉过来就可以了。看一下请求头一般就能找到。