目录
第一章：爬虫基础
1.1 什么是爬虫
爬虫（Web Crawler）是模拟浏览器访问网站，自动抓取数据的程序。
1.2 爬虫工作流程
1.3 爬虫伦理
1.4 环境准备
pip install requestspip install beautifulsoup4pip install lxmlpip install pandaspip install scrapy
第二章：网络请求
2.1 requests基础
import requests# GET请求response = requests.get("https://www.example.com")print(response.status_code) # 状态码print(response.text) # 文本内容print(response.content) # 字节内容print(response.headers) # 响应头print(response.url) # 请求URL# 带参数的GET请求params = {"page": 1, "size": 10}response = requests.get("https://api.example.com/data", params=params)# POST请求data = {"username": "test", "password": "123456"}response = requests.post("https://httpbin.org/post", data=data)# JSON请求json_data = {"name": "test", "age": 12}response = requests.post("https://httpbin.org/post", json=json_data)print(response.json())
2.2 请求头设置
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36","Referer": "https://www.example.com","Accept": "text/html,application/xhtml+xml","Accept-Language": "zh-CN,zh;q=0.9","Cookie": "session=abc123"}response = requests.get("https://www.example.com", headers=headers)
2.3 会话保持
session = requests.Session()# 登录login_data = {"username": "test", "password": "123456"}session.post("https://www.example.com/login", data=login_data)# 使用同一会话访问需要登录的页面response = session.get("https://www.example.com/profile")
2.4 超时与重试
from requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# 超时设置response = requests.get("https://www.example.com", timeout=5)# 重试机制session = requests.Session()retries = Retry(total=3, backoff_factor=1)adapter = HTTPAdapter(max_retries=retries)session.mount("http://", adapter)session.mount("https://", adapter)response = session.get("https://www.example.com")
2.5 SSL证书验证
# 禁用SSL验证（不推荐用于生产环境）response = requests.get("https://www.example.com", verify=False)# 使用自定义证书response = requests.get("https://www.example.com", cert="/path/to/cert.pem")
第三章：数据解析
3.1 BeautifulSoup
from bs4 import BeautifulSouphtml = """<html><head><title>测试页面</title></head><body> <div class="content"> <h1>标题</h1> <p class="text">段落1</p> <p class="text">段落2</p> <a href="https://example.com">链接</a> </div></body></html>"""soup = BeautifulSoup(html, "lxml")# 查找元素title = soup.title.stringprint(title) # 测试页面h1 = soup.h1.textprint(h1) # 标题# 查找所有段落paragraphs = soup.find_all("p")for p in paragraphs: print(p.text)# 按类名查找text_elements = soup.find_all("p", class_="text")# 按属性查找link = soup.find("a")print(link["href"]) # https://example.com# CSS选择器elements = soup.select(".content p")
3.2 lxml
from lxml import etreehtml = """<div class="content"> <h1>标题</h1> <p>段落1</p> <p>段落2</p></div>"""tree = etree.HTML(html)# XPath选择h1 = tree.xpath("//h1/text()")[0]print(h1) # 标题paragraphs = tree.xpath("//p/text()")for p in paragraphs: print(p)# 属性选择link = tree.xpath("//a/@href")print(link)
3.3 re正则表达式
import rehtml = '<a href="https://example.com">链接</a>'# 提取链接url = re.search(r'href="(.*?)"', html)if url: print(url.group(1)) # https://example.com# 提取所有链接urls = re.findall(r'href="(.*?)"', html)print(urls)
3.4 PyQuery
from pyquery import PyQuery as pqhtml = """<div class="content"> <h1>标题</h1> <p>段落1</p> <p>段落2</p></div>"""doc = pq(html)# jQuery风格选择h1 = doc("h1").text()print(h1) # 标题paragraphs = doc("p").items()for p in paragraphs: print(p.text())
第四章：反爬处理
4.1 User-Agent轮换
import randomuser_agents = ["Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36","Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36","Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"]headers = {"User-Agent": random.choice(user_agents)}response = requests.get("https://www.example.com", headers=headers)
4.2 IP代理
proxies = {"http": "http://127.0.0.1:7890","https": "http://127.0.0.1:7890"}response = requests.get("https://www.example.com", proxies=proxies)# 代理池proxy_list = ["http://proxy1:port","http://proxy2:port","http://proxy3:port"]proxy = {"http": random.choice(proxy_list)}response = requests.get("https://www.example.com", proxies=proxy)
4.3 验证码处理
# 打码平台API（示例）defrecognize_captcha(image_path):# 调用打码平台API# 返回识别结果pass# Selenium处理滑动验证码from selenium import webdriverfrom selenium.webdriver.common.action_chains import ActionChainsdriver = webdriver.Chrome()driver.get("https://www.example.com")# 定位滑块slider = driver.find_element_by_id("slider")# 模拟拖动action = ActionChains(driver)action.click_and_hold(slider).perform()action.move_by_offset(300, 0).perform()action.release().perform()
4.4 Cookie处理
# 从浏览器复制Cookiecookies = {"session": "abc123","user": "test"}response = requests.get("https://www.example.com", cookies=cookies)# 使用Session保持Cookiesession = requests.Session()response1 = session.get("https://www.example.com/login")response2 = session.get("https://www.example.com/profile")
4.5 JavaScript渲染
# Seleniumfrom selenium import webdriverfrom selenium.webdriver.common.by import Byfrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECdriver = webdriver.Chrome()driver.get("https://www.example.com")# 等待元素加载wait = WebDriverWait(driver, 10)element = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "content")))# 获取渲染后的页面html = driver.page_sourcedriver.quit()# Pyppeteer（异步）import asynciofrom pyppeteer import launchasyncdefscrape(): browser = await launch() page = await browser.newPage()await page.goto("https://www.example.com") content = await page.content()await browser.close()return contentcontent = asyncio.get_event_loop().run_until_complete(scrape())
第五章：数据存储
5.1 CSV文件
import csv# 写入CSVwith open("data.csv", "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["姓名", "年龄", "城市"]) writer.writerow(["小明", 12, "北京"]) writer.writerow(["小红", 14, "上海"])# 读取CSVwith open("data.csv", "r", encoding="utf-8") as f: reader = csv.reader(f)for row in reader: print(row)# 使用pandasimport pandas as pddf = pd.DataFrame({"姓名": ["小明", "小红"], "年龄": [12, 14]})df.to_csv("data.csv", index=False)
5.2 JSON文件
import json# 写入JSONdata = {"name": "小明", "age": 12, "scores": [85, 92, 78]}with open("data.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2)# 读取JSONwith open("data.json", "r", encoding="utf-8") as f: data = json.load(f) print(data)
5.3 MySQL数据库
import pymysqldefsave_to_mysql(data): conn = pymysql.connect( host="localhost", user="root", password="password", database="spider_db", charset="utf8mb4" ) cursor = conn.cursor()# 创建表 cursor.execute(""" CREATE TABLE IF NOT EXISTS articles ( id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255), url VARCHAR(255), content TEXT ) """)# 插入数据 sql = "INSERT INTO articles (title, url, content) VALUES (%s, %s, %s)" cursor.execute(sql, (data["title"], data["url"], data["content"])) conn.commit() conn.close()
5.4 MongoDB
from pymongo import MongoClientdefsave_to_mongodb(data): client = MongoClient("localhost", 27017) db = client["spider_db"] collection = db["articles"]# 插入单条数据 collection.insert_one(data)# 插入多条数据# collection.insert_many(data_list)# 查询数据 results = collection.find({"title": {"$regex": "Python"}})for result in results: print(result) client.close()
5.5 Redis缓存
import redisdefcache_url(url): r = redis.Redis(host="localhost", port=6379, db=0)# 设置缓存 r.set(url, "1", ex=3600) # 1小时过期# 检查是否已爬取if r.exists(url):returnTrue# 已爬取returnFalse# 未爬取
第六章：Scrapy框架
6.1 创建项目
scrapy startproject myspidercd myspiderscrapy genspider example example.com
6.2 项目结构
myspider/├── myspider/│ ├── __init__.py│ ├── items.py│ ├── middlewares.py│ ├── pipelines.py│ ├── settings.py│ └── spiders/│ ├── __init__.py│ └── example.py└── scrapy.cfg
6.3 定义Item
# items.pyimport scrapyclassArticleItem(scrapy.Item): title = scrapy.Field() url = scrapy.Field() content = scrapy.Field() publish_time = scrapy.Field()
6.4 编写Spider
# spiders/example.pyimport scrapyfrom myspider.items import ArticleItemclassExampleSpider(scrapy.Spider): name = "example" allowed_domains = ["example.com"] start_urls = ["https://www.example.com"]defparse(self, response):# 解析文章列表 articles = response.css(".article-item")for article in articles: item = ArticleItem() item["title"] = article.css("h2::text").get() item["url"] = article.css("a::attr(href)").get()# 跟进文章详情页yield scrapy.Request( url=response.urljoin(item["url"]), callback=self.parse_article, meta={"item": item} )# 翻页 next_page = response.css(".next-page::attr(href)").get()if next_page:yield scrapy.Request( url=response.urljoin(next_page), callback=self.parse )defparse_article(self, response): item = response.meta["item"] item["content"] = response.css(".content::text").getall() item["publish_time"] = response.css(".time::text").get()yield item
6.5 配置Settings
# settings.pyBOT_NAME = "myspider"SPIDER_MODULES = ["myspider.spiders"]NEWSPIDER_MODULE = "myspider.spiders"# 请求头DEFAULT_REQUEST_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)","Accept": "text/html,application/xhtml+xml",}# 下载延迟（秒）DOWNLOAD_DELAY = 1# 并发请求数CONCURRENT_REQUESTS = 16# 禁用CookieCOOKIES_ENABLED = False# 中间件DOWNLOADER_MIDDLEWARES = {"myspider.middlewares.UserAgentMiddleware": 400,}# 管道ITEM_PIPELINES = {"myspider.pipelines.DataPipeline": 300,}
6.6 中间件
# middlewares.pyimport randomclassUserAgentMiddleware:def__init__(self): self.user_agents = ["Mozilla/5.0 (Windows NT 10.0; Win64; x64)","Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)", ]defprocess_request(self, request, spider): request.headers["User-Agent"] = random.choice(self.user_agents)
6.7 Pipeline
# pipelines.pyimport jsonimport pymysqlclassDataPipeline:defopen_spider(self, spider): self.file = open("data.json", "w", encoding="utf-8")defclose_spider(self, spider): self.file.close()defprocess_item(self, item, spider): line = json.dumps(dict(item), ensure_ascii=False) + "\n" self.file.write(line)return itemclassMysqlPipeline:defopen_spider(self, spider): self.conn = pymysql.connect( host="localhost", user="root", password="password", database="spider_db" ) self.cursor = self.conn.cursor()defclose_spider(self, spider): self.conn.close()defprocess_item(self, item, spider): sql = "INSERT INTO articles (title, url, content) VALUES (%s, %s, %s)" self.cursor.execute(sql, (item["title"], item["url"], item["content"])) self.conn.commit()return item
6.8 运行爬虫
scrapy crawl example# 导出数据scrapy crawl example -o data.json
第七章：实战项目
7.1 爬取豆瓣电影Top250
import requestsfrom bs4 import BeautifulSoupimport csvimport timedefscrape_douban_top250(): headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } movies = []for page in range(10): url = f"https://movie.douban.com/top250?start={page * 25}" response = requests.get(url, headers=headers) soup = BeautifulSoup(response.text, "lxml") items = soup.find_all("div", class_="item")for item in items: movie = {} movie["rank"] = item.find("em").text movie["title"] = item.find("span", class_="title").text movie["rating"] = item.find("span", class_="rating_num").text movie["info"] = item.find("div", class_="bd").p.text.strip() movies.append(movie) print(f"已完成第{page + 1}页") time.sleep(1)# 保存到CSVwith open("douban_top250.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["rank", "title", "rating", "info"]) writer.writeheader() writer.writerows(movies) print(f"共抓取{len(movies)}部电影")if __name__ == "__main__": scrape_douban_top250()
7.2 爬取新闻网站
import requestsfrom bs4 import BeautifulSoupimport jsonimport timedefscrape_news(): headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } articles = []for page in range(1, 6): url = f"https://www.example.com/news?page={page}" response = requests.get(url, headers=headers) soup = BeautifulSoup(response.text, "lxml") items = soup.find_all("div", class_="news-item")for item in items: article = {} article["title"] = item.find("h2").text.strip() article["url"] = item.find("a")["href"] article["summary"] = item.find("p", class_="summary").text.strip() article["time"] = item.find("span", class_="time").text.strip() articles.append(article) print(f"已完成第{page}页") time.sleep(1)# 保存到JSONwith open("news.json", "w", encoding="utf-8") as f: json.dump(articles, f, ensure_ascii=False, indent=2) print(f"共抓取{len(articles)}篇文章")if __name__ == "__main__": scrape_news()
7.3 爬取商品信息
import requestsfrom bs4 import BeautifulSoupimport csvimport timedefscrape_products(): headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } products = []for page in range(1, 6): url = f"https://www.example.com/products?page={page}" response = requests.get(url, headers=headers) soup = BeautifulSoup(response.text, "lxml") items = soup.find_all("div", class_="product-item")for item in items: product = {} product["name"] = item.find("h3", class_="name").text.strip() product["price"] = item.find("span", class_="price").text.strip() product["sales"] = item.find("span", class_="sales").text.strip() product["shop"] = item.find("span", class_="shop").text.strip() products.append(product) print(f"已完成第{page}页") time.sleep(1)# 保存到CSVwith open("products.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["name", "price", "sales", "shop"]) writer.writeheader() writer.writerows(products) print(f"共抓取{len(products)}个商品")if __name__ == "__main__": scrape_products()
7.4 Scrapy实战：知乎爬虫
# zhihu/spiders/zhihu_spider.pyimport scrapyfrom zhihu.items import ZhihuItemclassZhihuSpider(scrapy.Spider): name = "zhihu" allowed_domains = ["zhihu.com"] custom_settings = {"DEFAULT_REQUEST_HEADERS": {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)","Referer": "https://www.zhihu.com", } } start_urls = ["https://www.zhihu.com/topics"]defparse(self, response):# 解析话题列表 topics = response.css(".zm-topic-list-item")for topic in topics: url = topic.css("a::attr(href)").get()yield scrapy.Request( url=response.urljoin(url), callback=self.parse_topic )defparse_topic(self, response):# 解析话题下的问题 questions = response.css(".QuestionItem-title a::attr(href)").getall()for question_url in questions:yield scrapy.Request( url=response.urljoin(question_url), callback=self.parse_question )defparse_question(self, response): item = ZhihuItem() item["title"] = response.css("h1::text").get() item["content"] = response.css(".RichContent-inner::text").getall() item["url"] = response.urlyield item
7.5 异步爬虫
import asyncioimport aiohttpfrom bs4 import BeautifulSoupasyncdeffetch(session, url):asyncwith session.get(url) as response:returnawait response.text()asyncdefparse(html): soup = BeautifulSoup(html, "lxml") titles = soup.find_all("h2")return [title.text for title in titles]asyncdefscrape_page(session, url): html = await fetch(session, url)returnawait parse(html)asyncdefmain(): urls = [f"https://www.example.com/page/{i}"for i in range(1, 11)]asyncwith aiohttp.ClientSession() as session: tasks = [scrape_page(session, url) for url in urls] results = await asyncio.gather(*tasks) all_titles = []for titles in results: all_titles.extend(titles) print(f"共抓取{len(all_titles)}个标题")if __name__ == "__main__": asyncio.run(main())
7.6 Selenium实战：动态页面
from selenium import webdriverfrom selenium.webdriver.common.by import Byfrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECimport timedefscrape_dynamic_page(): options = webdriver.ChromeOptions() options.add_argument("--headless") # 无头模式 driver = webdriver.Chrome(options=options) driver.get("https://www.example.com")# 等待元素加载 wait = WebDriverWait(driver, 10) element = wait.until( EC.presence_of_element_located((By.CLASS_NAME, "content")) )# 滚动加载更多内容for i in range(5): driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") time.sleep(1)# 提取数据 items = driver.find_elements(By.CLASS_NAME, "item")for item in items: title = item.find_element(By.TAG_NAME, "h2").text print(title) driver.quit()if __name__ == "__main__": scrape_dynamic_page()
附录：爬虫常用库
结语
爬虫技术应用广泛，但请遵守以下原则：
祝你爬虫学习顺利！
10个月宝宝每天需要喝多少奶粉？
