目录

1. 爬虫基础
2. 网络请求
3. 数据解析
4. 反爬处理
5. 数据存储
6. 框架爬虫
7. 实战项目

第一章：爬虫基础

1.1 什么是爬虫

爬虫（Web Crawler）是模拟浏览器访问网站，自动抓取数据的程序。

1.2 爬虫工作流程

1. 发送HTTP请求获取网页
2. 解析HTML提取数据
3. 存储数据到文件或数据库
4. 循环抓取下一页

1.3 爬虫伦理

• 遵守robots.txt协议
• 控制请求频率，避免给服务器造成压力
• 不抓取敏感数据
• 仅用于学习和合法用途

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()

附录：爬虫常用库

结语

爬虫技术应用广泛，但请遵守以下原则：

1. 遵守网站robots.txt协议
2. 控制爬取频率
3. 不侵犯他人隐私
4. 仅用于合法用途

祝你爬虫学习顺利！