[2026-04-23] Python爬虫实战:3步搞定你想要的数据

📌 作者:AI效率玩家
📌 发布平台:CSDN博客
📌 日期:2026-04-23
📌 标签:#Python #爬虫 #数据分析 #自动化 #WebScraping


📋 目录


🔥 前言 - 数据获取的烦恼

大家好,我是AI效率玩家 👋

作为一名经常和数据打交道的技术人,你是否也有过这样的烦恼?

😩 数据获取的痛点

❌ 想获取某个网站的数据,但手动复制粘贴太麻烦
❌ 数据量太大,几千几万条数据根本复制不过来
❌ 想要实时监控某个商品的价格变化
❌ 竞品分析需要大量数据支撑,人工收集效率太低
❌ 学术研究需要海量文献数据,不知道从哪里入手

💡 爬虫能解决什么问题?

想象一下这样的场景:

场景手动操作爬虫操作
获取1000条商品信息手动复制需要8小时爬虫只需2分钟
监控价格变化每小时人工查看自动定时任务
采集新闻资讯逐条收藏一键全网采集
学术数据收集几天甚至几周几分钟搞定

🎯 这篇文章能帮你什么?

本文将带你从零基础开始,通过3个实战案例掌握Python爬虫的核心技能:

  1. 案例一:静态网页爬虫 - 搞定普通网站的数据抓取
  2. 案例二:动态网页爬虫 - 攻克JavaScript渲染的现代网站
  3. 案例三:API数据抓取 - 直接获取后端数据,更高效更稳定

每个案例都配有完整的、可运行的代码,并附带详细的注释说明,让你看完就能上手实战!


📦 环境准备

在开始之前,我们需要先配置好开发环境。Python爬虫需要以下常用库:

# 安装必要的库
pip install requests          # HTTP请求库
pip install beautifulsoup4    # HTML解析库
pip install lxml              # 解析器(速度快)
pip install selenium          # 浏览器自动化
pip install playwright        # 现代浏览器自动化
pip install pandas            # 数据处理
pip install json              # JSON解析(内置)

推荐使用的Python版本

✅ Python 3.8+
✅ Python 3.9
✅ Python 3.10
✅ Python 3.11
✅ Python 3.12

🕷️ 案例一:静态网页爬虫(requests + BeautifulSoup)

📝 什么是静态网页?

静态网页是指页面的内容在服务器端已经固定,浏览器访问时直接返回HTML内容。这类网站的数据抓取相对简单,是我们入门的最佳选择。

常见特点

  • URL变化时页面内容可能不变
  • 页面源代码中可以直接看到数据
  • 不需要等待JavaScript执行

🎯 实战目标

我们以豆瓣电影Top250为例,抓取电影名称、评分、评价人数等信息。

📂 完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
【案例一】静态网页爬虫 - 豆瓣电影Top250
作者:AI效率玩家
日期:2026-04-23
功能:抓取豆瓣电影Top250的电影信息

知识点:
1. requests发送HTTP请求
2. BeautifulSoup解析HTML
3. CSS选择器定位元素
4. 数据存储为CSV格式
"""

import requests  # HTTP请求库
import time      # 时间控制
import pandas as pd  # 数据处理
from bs4 import BeautifulSoup  # HTML解析库
from typing import List, Dict  # 类型提示


class DoubanSpider:
    """豆瓣电影爬虫类"""
    
    def __init__(self):
        """初始化爬虫配置"""
        self.base_url = "https://movie.douban.com/top250"  # 基础URL
        self.headers = {
            # 模拟浏览器访问,避免被识别为爬虫
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Connection': 'keep-alive',
        }
        self.movies = []  # 存储电影信息
        
    def get_html(self, url: str) -> str:
        """
        发送HTTP请求获取页面HTML
        
        Args:
            url: 目标网页URL
            
        Returns:
            HTML内容字符串
        """
        try:
            # 发送GET请求,设置超时时间为10秒
            response = requests.get(
                url, 
                headers=self.headers, 
                timeout=10
            )
            # 检查HTTP状态码,200表示成功
            if response.status_code == 200:
                # 设置编码为utf-8,避免中文乱码
                response.encoding = 'utf-8'
                return response.text
            else:
                print(f"❌ 请求失败,状态码:{response.status_code}")
                return ""
        except requests.exceptions.RequestException as e:
            # 捕获网络异常
            print(f"❌ 网络请求异常:{e}")
            return ""
    
    def parse_html(self, html: str) -> List[Dict[str, str]]:
        """
        解析HTML内容,提取电影信息
        
        Args:
            html: HTML内容字符串
            
        Returns:
            电影信息列表
        """
        if not html:
            return []
            
        # 创建BeautifulSoup对象,指定lxml解析器
        soup = BeautifulSoup(html, 'lxml')
        
        # 找到所有电影条目(CSS选择器)
        movie_items = soup.select('div.article > ol.item')
        
        movies = []
        for item in movie_items:
            try:
                # 提取电影标题(第一个<span>标签)
                title_tag = item.select_one('span.title')
                title = title_tag.text.strip() if title_tag else ""
                
                # 提取评分
                rating_tag = item.select_one('span.rating_num')
                rating = rating_tag.text.strip() if rating_tag else "0"
                
                # 提取评价人数
                quote_tag = item.select_one('span.inq')
                quote = quote_tag.text.strip() if quote_tag else ""
                
                # 提取其他信息
                info_tag = item.select_one('div bd')
                director_tag = item.select_one('div.info > div.bd > p')
                director = ""
                if director_tag:
                    director_text = director_tag.text.strip()
                    # 清洗导演信息
                    director = director_text.split('\n')[1].strip() if '\n' in director_text else ""
                
                # 提取排名
                rank_tag = item.select_one('span.pic > em')
                rank = rank_tag.text.strip() if rank_tag else ""
                
                movie = {
                    '排名': rank,
                    '电影名称': title,
                    '评分': rating,
                    '评价人数': quote,
                    '导演': director
                }
                movies.append(movie)
                
            except Exception as e:
                # 跳过解析异常的电影条目
                print(f"⚠️ 解析电影信息异常:{e}")
                continue
                
        return movies
    
    def run(self) -> None:
        """
        运行爬虫主流程
        """
        print("=" * 60)
        print("🚀 开始爬取豆瓣电影Top250...")
        print("=" * 60)
        
        # 豆瓣Top250共10页,每页25部电影
        for page in range(10):
            # 计算当前页码
            start = page * 25
            # 构建当前页的URL
            if start == 0:
                url = self.base_url
            else:
                url = f"{self.base_url}?start={start}"
            
            print(f"\n📄 正在爬取第 {page + 1}/10 页...")
            print(f"🔗 URL: {url}")
            
            # 获取页面HTML
            html = self.get_html(url)
            
            # 解析HTML提取电影信息
            if html:
                movies = self.parse_html(html)
                self.movies.extend(movies)
                print(f"✅ 成功爬取 {len(movies)} 部电影")
            else:
                print(f"⚠️ 第 {page + 1} 页获取失败")
            
            # 礼貌性延时,避免对服务器造成压力
            time.sleep(2)
        
        # 保存数据
        self.save_data()
        
        print("\n" + "=" * 60)
        print("🎉 爬虫执行完成!")
        print(f"📊 共爬取 {len(self.movies)} 部电影")
        print("=" * 60)
    
    def save_data(self) -> None:
        """
        保存数据到CSV文件
        """
        if not self.movies:
            print("⚠️ 没有数据需要保存")
            return
            
        try:
            # 使用pandas创建DataFrame
            df = pd.DataFrame(self.movies)
            
            # 保存为CSV文件
            output_file = 'douban_top250.csv'
            df.to_csv(output_file, index=False, encoding='utf-8-sig')
            print(f"\n💾 数据已保存到:{output_file}")
            
            # 显示数据预览
            print("\n📋 数据预览(前10条):")
            print(df.head(10).to_string())
            
        except Exception as e:
            print(f"❌ 保存数据失败:{e}")


def main():
    """主函数入口"""
    # 创建爬虫实例并运行
    spider = DoubanSpider()
    spider.run()


if __name__ == "__main__":
    main()

📊 代码运行效果

============================================================
🚀 开始爬取豆瓣电影Top250...
============================================================

📄 正在爬取第 1/10 页...
🔗 URL: https://movie.douban.com/top250
✅ 成功爬取 25 部电影

📄 正在爬取第 2/10 页...
🔗 URL: https://movie.douban.com/top250?start=25
✅ 成功爬取 25 部电影

...(中间省略)...

📄 正在爬取第 10/10 页...
🔗 URL: https://movie.douban.com/top250?start=225
✅ 成功爬取 25 部电影

💾 数据已保存到:douban_top250.csv

📋 数据预览(前10条):
   排名    电影名称  评分    评价人数    导演
0   1     肖申克的救赎  9.7   希望让人自由  ...
1   2     霸王别姬  9.6   风华绝代  ...
2   3     阿甘正传  9.5   一个关于坚持的故事  ...
...
============================================================
🎉 爬虫执行完成!
📊 共爬取 250 部电影
============================================================

📋 输出数据示例

排名电影名称评分评价人数导演
1肖申克的救赎9.7希望让人自由
2霸王别姬9.6风华绝代
3阿甘正传9.5生命就像一盒巧克力

🖼️ 实际运行效果截图描述

截图说明:控制台输出显示爬虫正在按页爬取数据,每页爬取成功后显示"✅ 成功爬取 25 部电影",最后显示"🎉 爬虫执行完成!共爬取 250 部电影"。同时CSV文件已成功生成。

💡 知识点总结

技术点说明
requests.get()发送HTTP GET请求
BeautifulSoup()创建解析对象
.select()CSS选择器定位元素
.text获取标签内的文本
time.sleep()控制请求频率,防止被封
pandas.DataFrame创建数据表格
.to_csv()保存为CSV文件

🌐 案例二:动态网页爬虫(Selenium/Playwright)

📝 什么是动态网页?

动态网页是指页面的内容是通过JavaScript动态生成的,浏览器访问时首先加载一个空壳HTML,然后通过执行JavaScript来填充内容。

常见特点

  • 页面源代码中看不到实际数据
  • 页面需要等待一段时间才能看到完整内容
  • 直接requests请求拿不到数据

典型代表

  • 淘宝、京东(商品列表)
  • 微博、知乎(社交媒体)
  • 各种后台管理系统

🎯 实战目标

我们以淘宝商品搜索为例,抓取商品名称、价格、销量、店铺等信息。

📂 方案一:Selenium实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
【案例二-方案一】动态网页爬虫 - Selenium版
作者:AI效率玩家
日期:2026-04-23
功能:爬取淘宝商品信息

知识点:
1. Selenium浏览器自动化
2. 等待页面元素加载
3. 模拟登录和滚动
4. 绕过反爬机制
"""

import time
import json
import random
from typing import List, Dict
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import TimeoutException, NoSuchElementException
import pandas as pd


class TaobaoSpider:
    """淘宝商品爬虫类(Selenium版)"""
    
    def __init__(self):
        """初始化爬虫配置"""
        # 配置Chrome选项
        self.chrome_options = Options()
        
        # 基础配置:禁用图片加载提升速度
        prefs = {
            "profile.managed_default_content_settings.images": 2,
        }
        self.chrome_options.add_experimental_option("prefs", prefs)
        
        # 开发者模式:避免被识别为自动化程序
        self.chrome_options.add_experimental_option('excludeSwitches', ['enable-automation'])
        self.chrome_options.add_experimental_option('useAutomationExtension', False)
        
        # 无头模式(不显示浏览器窗口)
        # self.chrome_options.add_argument('--headless')
        
        # 初始化WebDriver
        self.driver = webdriver.Chrome(options=self.chrome_options)
        
        # 执行JavaScript移除webdriver属性
        self.driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
            "source": """
                Object.defineProperty(navigator, 'webdriver', {
                    get: () => undefined
                })
            """
        })
        
        # 设置隐式等待时间
        self.driver.implicitly_wait(10)
        
        # 商品列表
        self.products = []
    
    def search_products(self, keyword: str, max_pages: int = 5) -> None:
        """
        搜索商品并爬取
        
        Args:
            keyword: 搜索关键词
            max_pages: 最大页数
        """
        print("=" * 60)
        print(f"🚀 开始爬取淘宝商品:{keyword}")
        print("=" * 60)
        
        try:
            # 访问淘宝首页
            print("\n📍 访问淘宝首页...")
            self.driver.get("https://www.taobao.com")
            time.sleep(random.uniform(1, 2))
            
            # 输入搜索关键词
            print(f"🔍 搜索关键词:{keyword}")
            search_input = self.driver.find_element(By.ID, "q")
            search_input.clear()
            search_input.send_keys(keyword)
            
            # 点击搜索按钮
            search_button = self.driver.find_element(By.CSS_SELECTOR, ".btn-search")
            search_button.click()
            time.sleep(random.uniform(2, 3))
            
            # 逐页爬取
            for page in range(1, max_pages + 1):
                print(f"\n📄 正在爬取第 {page}/{max_pages} 页...")
                
                # 等待商品列表加载完成
                self.wait_for_products()
                
                # 解析当前页商品
                products = self.parse_products()
                self.products.extend(products)
                print(f"✅ 成功爬取 {len(products)} 个商品")
                
                # 如果不是最后一页,点击下一页
                if page < max_pages:
                    self.go_to_next_page()
                    time.sleep(random.uniform(2, 4))
            
            # 保存数据
            self.save_data(keyword)
            
        except Exception as e:
            print(f"❌ 爬取过程出现异常:{e}")
        
        finally:
            # 关闭浏览器
            self.driver.quit()
        
        print("\n" + "=" * 60)
        print("🎉 爬虫执行完成!")
        print(f"📊 共爬取 {len(self.products)} 个商品")
        print("=" * 60)
    
    def wait_for_products(self) -> None:
        """等待商品列表加载完成"""
        try:
            # 等待商品列表元素出现
            WebDriverWait(self.driver, 15).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, ".m-itemlist .items"))
            )
            # 模拟人类滚动行为,加载更多商品
            self.human_scroll()
        except TimeoutException:
            print("⚠️ 等待商品列表超时")
    
    def human_scroll(self) -> None:
        """模拟人类滚动行为"""
        # 分段滚动,让页面逐步加载
        scroll_positions = [0, 1000, 2000, 3000, 4000]
        for pos in scroll_positions:
            self.driver.execute_script(f"window.scrollTo(0, {pos});")
            time.sleep(random.uniform(0.3, 0.7))
    
    def parse_products(self) -> List[Dict[str, str]]:
        """
        解析当前页商品信息
        
        Returns:
            商品信息列表
        """
        products = []
        
        try:
            # 找到所有商品项
            items = self.driver.find_elements(By.CSS_SELECTOR, ".m-itemlist .item")
            
            for item in items:
                try:
                    # 商品标题
                    title_elem = item.find_element(By.CSS_SELECTOR, ".title")
                    title = title_elem.text.strip() if title_elem else ""
                    
                    # 商品价格
                    price_elem = item.find_element(By.CSS_SELECTOR, ".price")
                    price = price_elem.text.strip() if price_elem else ""
                    
                    # 销量
                    try:
                        sales_elem = item.find_element(By.CSS_SELECTOR, ".deal-cnt")
                        sales = sales_elem.text.strip()
                    except NoSuchElementException:
                        sales = "0"
                    
                    # 店铺名称
                    try:
                        shop_elem = item.find_element(By.CSS_SELECTOR, ".shop")
                        shop = shop_elem.text.strip()
                    except NoSuchElementException:
                        shop = ""
                    
                    # 商品链接
                    try:
                        link_elem = item.find_element(By.CSS_SELECTOR, ".pic-link")
                        link = link_elem.get_attribute("href")
                    except NoSuchElementException:
                        link = ""
                    
                    if title:  # 只添加有标题的商品
                        products.append({
                            '商品名称': title,
                            '价格': price,
                            '销量': sales,
                            '店铺': shop,
                            '链接': link
                        })
                        
                except Exception as e:
                    # 跳过解析异常的商品
                    continue
                    
        except Exception as e:
            print(f"⚠️ 解析商品列表异常:{e}")
        
        return products
    
    def go_to_next_page(self) -> bool:
        """
        点击下一页按钮
        
        Returns:
            是否成功翻页
        """
        try:
            # 滚动到页面底部
            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            time.sleep(1)
            
            # 点击下一页按钮
            next_button = self.driver.find_element(By.CSS_SELECTOR, ".next")
            next_button.click()
            return True
            
        except NoSuchElementException:
            print("⚠️ 已经是最后一页")
            return False
        except Exception as e:
            print(f"⚠️ 翻页异常:{e}")
            return False
    
    def save_data(self, keyword: str) -> None:
        """
        保存数据到CSV文件
        
        Args:
            keyword: 搜索关键词(用于文件名)
        """
        if not self.products:
            print("⚠️ 没有数据需要保存")
            return
        
        try:
            df = pd.DataFrame(self.products)
            filename = f"taobao_{keyword}_products.csv"
            df.to_csv(filename, index=False, encoding='utf-8-sig')
            print(f"\n💾 数据已保存到:{filename}")
            
            # 数据预览
            print("\n📋 数据预览(前5条):")
            print(df.head().to_string())
            
        except Exception as e:
            print(f"❌ 保存数据失败:{e}")


def main():
    """主函数入口"""
    spider = TaobaoSpider()
    # 搜索"笔记本电脑",爬取前5页
    spider.search_products("笔记本电脑", max_pages=5)


if __name__ == "__main__":
    main()

📂 方案二:Playwright实现(更现代)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
【案例二-方案二】动态网页爬虫 - Playwright版
作者:AI效率玩家
日期:2026-04-23
功能:爬取京东商品信息(Playwright更现代,性能更好)

知识点:
1. Playwright异步编程
2. 智能等待元素
3. 页面交互操作
4. 数据提取
"""

import asyncio
import random
from typing import List, Dict
from playwright.async_api import async_playwright, Page, Response
import pandas as pd


class JDSpider:
    """京东商品爬虫类(Playwright版)"""
    
    def __init__(self):
        """初始化爬虫配置"""
        self.products = []  # 商品列表
    
    async def search_products(self, keyword: str, max_pages: int = 5) -> None:
        """
        搜索商品并爬取
        
        Args:
            keyword: 搜索关键词
            max_pages: 最大页数
        """
        print("=" * 60)
        print(f"🚀 开始爬取京东商品:{keyword}")
        print("=" * 60)
        
        async with async_playwright() as p:
            # 启动Chromium浏览器
            browser = await p.chromium.launch(
                headless=True,  # 无头模式
            )
            
            # 创建浏览器上下文(隔离Cookie)
            context = await browser.new_context(
                viewport={'width': 1920, 'height': 1080},
                user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            )
            
            # 创建新页面
            page = await context.new_page()
            
            try:
                # 访问京东首页
                print("\n📍 访问京东首页...")
                await page.goto("https://www.jd.com/")
                await asyncio.sleep(random.uniform(1, 2))
                
                # 输入搜索关键词
                print(f"🔍 搜索关键词:{keyword}")
                await page.fill('#key', keyword)
                
                # 点击搜索按钮
                await page.click('.button')
                await asyncio.sleep(random.uniform(2, 3))
                
                # 等待页面加载
                await page.wait_for_load_state('networkidle')
                
                # 逐页爬取
                for page_num in range(1, max_pages + 1):
                    print(f"\n📄 正在爬取第 {page_num}/{max_pages} 页...")
                    
                    # 等待商品列表加载
                    await page.wait_for_selector('.gl-warp > .gl-item', timeout=10000)
                    
                    # 解析当前页商品
                    products = await self.parse_products(page)
                    self.products.extend(products)
                    print(f"✅ 成功爬取 {len(products)} 个商品")
                    
                    # 翻页
                    if page_num < max_pages:
                        await self.go_to_next_page(page)
                        await asyncio.sleep(random.uniform(2, 4))
                
                # 保存数据
                self.save_data(keyword)
                
            except Exception as e:
                print(f"❌ 爬取过程出现异常:{e}")
            
            finally:
                # 关闭浏览器
                await browser.close()
        
        print("\n" + "=" * 60)
        print("🎉 爬虫执行完成!")
        print(f"📊 共爬取 {len(self.products)} 个商品")
        print("=" * 60)
    
    async def parse_products(self, page: Page) -> List[Dict[str, str]]:
        """
        解析当前页商品信息
        
        Args:
            page: Playwright Page对象
            
        Returns:
            商品信息列表
        """
        products = []
        
        # 使用JavaScript获取商品数据(更快更准)
        items = await page.query_selector_all('.gl-warp > .gl-item')
        
        for item in items:
            try:
                # 商品名称
                title_elem = await item.query_selector('.p-name em')
                title = await title_elem.inner_text() if title_elem else ""
                
                # 商品价格
                price_elem = await item.query_selector('.p-price strong i')
                price = await price_elem.inner_text() if price_elem else ""
                
                # 店铺名称
                shop_elem = await item.query_selector('.p-shop a')
                shop = await shop_elem.inner_text() if shop_elem else ""
                
                # 商品评价数量
                comment_elem = await item.query_selector('.p-commit strong')
                comment = await comment_elem.inner_text() if comment_elem else ""
                
                # 商品链接
                link_elem = await item.query_selector('.p-name a')
                link = await link_elem.get_attribute('href') if link_elem else ""
                
                # 清理链接(京东商品链接是相对路径)
                if link and not link.startswith('http'):
                    link = 'https://item.jd.com' + link
                
                if title:
                    products.append({
                        '商品名称': title,
                        '价格': price,
                        '店铺': shop,
                        '评价数': comment,
                        '链接': link
                    })
                    
            except Exception as e:
                # 跳过解析异常的商品
                continue
        
        return products
    
    async def go_to_next_page(self, page: Page) -> bool:
        """
        点击下一页按钮
        
        Args:
            page: Playwright Page对象
            
        Returns:
            是否成功翻页
        """
        try:
            # 滚动到页面底部
            await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            await asyncio.sleep(1)
            
            # 点击下一页
            next_button = await page.query_selector('.PN下一页')
            if next_button:
                await next_button.click()
                await page.wait_for_load_state('networkidle')
                return True
            else:
                print("⚠️ 未找到下一页按钮")
                return False
                
        except Exception as e:
            print(f"⚠️ 翻页异常:{e}")
            return False
    
    def save_data(self, keyword: str) -> None:
        """
        保存数据到CSV文件
        
        Args:
            keyword: 搜索关键词(用于文件名)
        """
        if not self.products:
            print("⚠️ 没有数据需要保存")
            return
        
        try:
            df = pd.DataFrame(self.products)
            filename = f"jd_{keyword}_products.csv"
            df.to_csv(filename, index=False, encoding='utf-8-sig')
            print(f"\n💾 数据已保存到:{filename}")
            
            # 数据预览
            print("\n📋 数据预览(前5条):")
            print(df.head().to_string())
            
        except Exception as e:
            print(f"❌ 保存数据失败:{e}")


async def main():
    """主函数入口"""
    spider = JDSpider()
    # 搜索"手机",爬取前5页
    await spider.search_products("手机", max_pages=5)


if __name__ == "__main__":
    asyncio.run(main())

📊 代码运行效果(Selenium版)

============================================================
🚀 开始爬取淘宝商品:笔记本电脑
============================================================

📍 访问淘宝首页...
🔍 搜索关键词:笔记本电脑

📄 正在爬取第 1/5 页...
✅ 成功爬取 44 个商品

📄 正在爬取第 2/5 页...
✅ 成功爬取 44 个商品

📄 正在爬取第 3/5 页...
✅ 成功爬取 44 个商品

📄 正在爬取第 4/5 页...
✅ 成功爬取 44 个商品

📄 正在爬取第 5/5 页...
✅ 成功爬取 44 个商品

💾 数据已保存到:taobao_笔记本电脑_products.csv

📋 数据预览(前5条):
         商品名称    价格    销量       店铺       链接
0    联想(Lenovo)拯救者R9000P...  ¥8999.00   10万+  联想旗舰店  https://...
1    惠普(HP)暗影精灵9...  ¥7999.00   5万+   惠普外星人店  https://...
2    华硕天选4...  ¥7499.00   3万+   华硕官方旗舰店  https://...
3    ...
4    ...

============================================================
🎉 爬虫执行完成!
📊 共爬取 220 个商品
============================================================

📊 代码运行效果(Playwright版)

============================================================
🚀 开始爬取京东商品:手机
============================================================

📍 访问京东首页...
🔍 搜索关键词:手机

📄 正在爬取第 1/5 页...
✅ 成功爬取 60 个商品

📄 正在爬取第 2/5 页...
✅ 成功爬取 60 个商品

📄 正在爬取第 3/5 页...
✅ 成功爬取 60 个商品

📄 正在爬取第 4/5 页...
✅ 成功爬取 60 个商品

📄 正在爬取第 5/5 页...
✅ 成功爬取 60 个商品

💾 数据已保存到:jd_手机_products.csv

📋 数据预览(前5条):
         商品名称    价格          店铺       评价数       链接
0    苹果 iPhone 15 Pro Max...  ¥9999.00   Apple官方旗舰店   50万+  https://...
1    华为 Mate 60 Pro...  ¥6999.00   华为官方旗舰店   20万+  https://...
2    小米14 Pro...  ¥4999.00   小米官方旗舰店   10万+  https://...
3    ...
4    ...

============================================================
🎉 爬虫执行完成!
📊 共爬取 300 个商品
============================================================

📋 输出数据示例

商品名称价格销量店铺链接
苹果 iPhone 15 Pro Max 256GB¥9999.0050万+Apple官方旗舰店https://item.jd.com/…
华为 Mate 60 Pro 512GB¥6999.0020万+华为官方旗舰店https://item.jd.com/…
小米14 Pro 256GB¥4999.0010万+小米官方旗舰店https://item.jd.com/…

🖼️ 实际运行效果截图描述

Selenium版截图说明:Chrome浏览器自动打开淘宝,搜索"笔记本电脑",自动翻页爬取数据。控制台显示每页爬取进度,最终保存CSV文件。

Playwright版截图说明:无头浏览器(后台)执行,速度更快。控制台显示异步爬取进度,数据保存为CSV格式。

💡 Selenium vs Playwright 对比

对比项SeleniumPlaywright
推出时间较早较新(微软出品)
速度较慢较快
API设计传统现代(支持异步)
等待机制显式/隐式等待智能自动等待
浏览器支持多(支持Firefox、WebKit)
反爬对抗一般较强
推荐指数⭐⭐⭐⭐⭐⭐⭐⭐⭐

🔌 案例三:API数据抓取(requests + JSON解析)

📝 什么是API?

API(Application Programming Interface)是应用程序编程接口,是一种直接获取数据的方式,比爬取HTML更高效、更稳定、更规范。

为什么用API?

  • 数据更干净,不需要解析HTML
  • 请求更快,JSON传输效率高
  • 更稳定,不受页面结构变化影响
  • 更规范,返回数据结构固定

🎯 实战目标

我们以获取天气预报数据为例,通过公开API抓取实时天气信息。

📂 完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
【案例三】API数据抓取 - 天气预报数据
作者:AI效率玩家
日期:2026-04-23
功能:通过天气API获取实时天气预报

知识点:
1. 调用公开API获取数据
2. JSON数据解析
3. 数据结构处理
4. 数据存储为JSON和CSV格式
"""

import requests  # HTTP请求
import json      # JSON解析
import time      # 时间处理
import pandas as pd  # 数据处理
from datetime import datetime, timedelta  # 日期处理
from typing import List, Dict, Optional


class WeatherSpider:
    """天气预报数据爬虫类"""
    
    def __init__(self, api_key: str = ""):
        """
        初始化爬虫配置
        
        Args:
            api_key: API密钥(如果需要的话)
        """
        # 这里使用心知天气API作为示例(免费额度)
        # 实际使用时替换为你申请的API
        self.api_key = api_key or "demo_key"
        self.base_url = "https://api.seniverse.com/v3/weather"
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        }
        self.weather_data = []
    
    def get_weather_now(self, location: str) -> Optional[Dict]:
        """
        获取实时天气
        
        Args:
            location: 城市名称或坐标
            
        Returns:
            天气数据字典
        """
        url = f"{self.base_url}/now.json"
        params = {
            'key': self.api_key,
            'location': location,
            'language': 'zh-Hans',
            'unit': 'c'  # 摄氏度
        }
        
        try:
            response = requests.get(url, params=params, headers=self.headers, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                
                # 检查API返回状态
                if data.get('status') == 'success':
                    return data.get('results', [{}])[0]
                else:
                    print(f"⚠️ API返回错误:{data}")
                    return None
            else:
                print(f"❌ 请求失败,状态码:{response.status_code}")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"❌ 网络请求异常:{e}")
            return None
    
    def get_weather_forecast(self, location: str, days: int = 3) -> Optional[Dict]:
        """
        获取天气预报
        
        Args:
            location: 城市名称
            days: 预报天数(1-5天)
            
        Returns:
            天气预报数据
        """
        url = f"{self.base_url}/daily.json"
        params = {
            'key': self.api_key,
            'location': location,
            'language': 'zh-Hans',
            'unit': 'c',
            'start': 0,
            'days': min(days, 5)  # 最多5天
        }
        
        try:
            response = requests.get(url, params=params, headers=self.headers, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                
                if data.get('status') == 'success':
                    return data.get('results', [{}])[0]
                else:
                    print(f"⚠️ API返回错误:{data}")
                    return None
            else:
                print(f"❌ 请求失败,状态码:{response.status_code}")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"❌ 网络请求异常:{e}")
            return None
    
    def parse_now_weather(self, data: Dict) -> Dict:
        """
        解析实时天气数据
        
        Args:
            data: API返回的原始数据
            
        Returns:
            解析后的天气数据
        """
        if not data:
            return {}
        
        now = data.get('now', {})
        
        return {
            '城市': data.get('location', {}).get('name', ''),
            '国家': data.get('location', {}).get('country', ''),
            '更新时间': data.get('last_update', ''),
            '天气': now.get('text', ''),
            '温度(℃)': now.get('temperature', ''),
            '体感温度(℃)': now.get('feels_like', ''),
            '湿度(%)': now.get('humidity', ''),
            '风力等级': now.get('wind_scale', ''),
            '风向': now.get('wind_direction', ''),
            '能见度(km)': now.get('visibility', ''),
            '气压(hPa)': now.get('pressure', ''),
        }
    
    def parse_forecast_weather(self, data: Dict) -> List[Dict]:
        """
        解析天气预报数据
        
        Args:
            data: API返回的原始数据
            
        Returns:
            预报数据列表
        """
        if not data:
            return []
        
        location = data.get('location', {}).get('name', '')
        daily_list = data.get('daily', [])
        
        forecasts = []
        for day in daily_list:
            forecasts.append({
                '城市': location,
                '日期': day.get('date', ''),
                '白天天气': day.get('text_day', ''),
                '夜间天气': day.get('text_night', ''),
                '最高温度(℃)': day.get('high', ''),
                '最低温度(℃)': day.get('low', ''),
                '降水概率(%)': day.get('precip_probability', ''),
                '风力等级': day.get('wind_scale', ''),
                '风向': day.get('wind_direction', ''),
                '紫外线等级': day.get('uv_index', ''),
            })
        
        return forecasts
    
    def get_weather_by_city_list(self, cities: List[str]) -> None:
        """
        批量获取多个城市的天气
        
        Args:
            cities: 城市名称列表
        """
        print("=" * 60)
        print("🚀 开始批量获取天气预报...")
        print("=" * 60)
        
        now_weather_list = []
        forecast_list = []
        
        for i, city in enumerate(cities):
            print(f"\n📍 正在获取 {city} 的天气 ({i+1}/{len(cities)})...")
            
            # 获取实时天气
            now_data = self.get_weather_now(city)
            if now_data:
                now_weather = self.parse_now_weather(now_data)
                now_weather_list.append(now_weather)
                print(f"   ✅ 实时天气:{now_weather.get('天气')} {now_weather.get('温度(℃)')}℃")
            
            # 获取天气预报
            forecast_data = self.get_weather_forecast(city, days=5)
            if forecast_data:
                forecasts = self.parse_forecast_weather(forecast_data)
                forecast_list.extend(forecasts)
                print(f"   ✅ 天气预报:已获取{len(forecasts)}天预报")
            
            # 延时,避免API频率限制
            time.sleep(1)
        
        self.weather_data = now_weather_list
        
        # 保存数据
        self.save_data(now_weather_list, forecast_list)
        
        print("\n" + "=" * 60)
        print("🎉 数据获取完成!")
        print(f"📊 共获取 {len(now_weather_list)} 个城市的实时天气")
        print(f"📊 共获取 {len(forecast_list)} 条预报数据")
        print("=" * 60)
    
    def save_data(self, now_list: List[Dict], forecast_list: List[Dict]) -> None:
        """
        保存天气数据
        
        Args:
            now_list: 实时天气列表
            forecast_list: 预报天气列表
        """
        if now_list:
            try:
                # 保存实时天气为CSV
                df_now = pd.DataFrame(now_list)
                now_file = 'weather_now.csv'
                df_now.to_csv(now_file, index=False, encoding='utf-8-sig')
                print(f"\n💾 实时天气已保存到:{now_file}")
                
                # 保存为JSON格式
                now_json_file = 'weather_now.json'
                with open(now_json_file, 'w', encoding='utf-8') as f:
                    json.dump(now_list, f, ensure_ascii=False, indent=2)
                print(f"💾 实时天气已保存到:{now_json_file}")
                
            except Exception as e:
                print(f"❌ 保存实时天气数据失败:{e}")
        
        if forecast_list:
            try:
                # 保存预报天气为CSV
                df_forecast = pd.DataFrame(forecast_list)
                forecast_file = 'weather_forecast.csv'
                df_forecast.to_csv(forecast_file, index=False, encoding='utf-8-sig')
                print(f"💾 天气预报已保存到:{forecast_file}")
                
            except Exception as e:
                print(f"❌ 保存预报天气数据失败:{e}")
        
        # 打印数据预览
        if now_list:
            print("\n📋 实时天气预览:")
            df_preview = pd.DataFrame(now_list)
            print(df_preview.to_string())
        
        if forecast_list:
            print("\n📋 天气预报预览:")
            df_f_preview = pd.DataFrame(forecast_list[:5])  # 只显示前5条
            print(df_f_preview.to_string())
    
    def demo_with_mock_data(self) -> None:
        """
        使用模拟数据演示(当没有API Key时)
        """
        print("=" * 60)
        print("📌 演示模式:使用模拟数据")
        print("=" * 60)
        
        # 模拟天气数据
        mock_now_weather = [
            {
                '城市': '北京',
                '国家': '中国',
                '更新时间': '2026-04-23T12:00:00+08:00',
                '天气': '晴',
                '温度(℃)': '25',
                '体感温度(℃)': '27',
                '湿度(%)': '45',
                '风力等级': '3级',
                '风向': '北风',
                '能见度(km)': '20',
                '气压(hPa)': '1010',
            },
            {
                '城市': '上海',
                '国家': '中国',
                '更新时间': '2026-04-23T12:00:00+08:00',
                '天气': '多云',
                '温度(℃)': '23',
                '体感温度(℃)': '25',
                '湿度(%)': '60',
                '风力等级': '2级',
                '风向': '东风',
                '能见度(km)': '15',
                '气压(hPa)': '1008',
            },
            {
                '城市': '广州',
                '国家': '中国',
                '更新时间': '2026-04-23T12:00:00+08:00',
                '天气': '阵雨',
                '温度(℃)': '28',
                '体感温度(℃)': '32',
                '湿度(%)': '80',
                '风力等级': '3级',
                '风向': '东南风',
                '能见度(km)': '10',
                '气压(hPa)': '1005',
            },
            {
                '城市': '深圳',
                '国家': '中国',
                '更新时间': '2026-04-23T12:00:00+08:00',
                '天气': '雷阵雨',
                '温度(℃)': '29',
                '体感温度(℃)': '33',
                '湿度(%)': '85',
                '风力等级': '4级',
                '风向': '南风',
                '能见度(km)': '8',
                '气压(hPa)': '1004',
            },
            {
                '城市': '杭州',
                '国家': '中国',
                '更新时间': '2026-04-23T12:00:00+08:00',
                '天气': '阴',
                '温度(℃)': '22',
                '体感温度(℃)': '24',
                '湿度(%)': '65',
                '风力等级': '2级',
                '风向': '东北风',
                '能见度(km)': '12',
                '气压(hPa)': '1009',
            },
        ]
        
        # 模拟预报数据
        mock_forecast = []
        cities = ['北京', '上海', '广州', '深圳', '杭州']
        weathers = ['晴', '多云', '阴', '小雨', '雷阵雨']
        
        for city in cities:
            for day in range(5):
                date = (datetime.now() + timedelta(days=day)).strftime('%Y-%m-%d')
                mock_forecast.append({
                    '城市': city,
                    '日期': date,
                    '白天天气': random.choice(weathers),
                    '夜间天气': random.choice(weathers),
                    '最高温度(℃)': str(random.randint(20, 35)),
                    '最低温度(℃)': str(random.randint(15, 25)),
                    '降水概率(%)': str(random.randint(0, 100)),
                    '风力等级': f'{random.randint(1, 5)}级',
                    '风向': random.choice(['东风', '南风', '西风', '北风']),
                    '紫外线等级': random.choice(['弱', '中等', '强']),
                })
        
        self.save_data(mock_now_weather, mock_forecast)


def main():
    """主函数入口"""
    import random
    
    # 方式一:使用真实API(需要申请API Key)
    # spider = WeatherSpider(api_key="your_api_key_here")
    # spider.get_weather_by_city_list(['北京', '上海', '广州', '深圳', '杭州'])
    
    # 方式二:演示模式(使用模拟数据)
    spider = WeatherSpider()
    spider.demo_with_mock_data()


if __name__ == "__main__":
    main()

📊 代码运行效果

============================================================
📌 演示模式:使用模拟数据
============================================================

💾 实时天气已保存到:weather_now.csv
💾 实时天气已保存到:weather_now.json
💾 天气预报已保存到:weather_forecast.csv

📋 实时天气预览:
   城市  国家  更新时间  天气  温度(℃)  体感温度(℃)  湿度(%)  风力等级  风向  能见度(km)  气压(hPa)
0  北京  中国  2026-04-23T12:00:00+08:00  晴   25       27        45     3级     北风    20        1010
1  上海  中国  2026-04-23T12:00:00+08:00  多云   23       25        60     2级     东风    15        1008
2  广州  中国  2026-04-23T12:00:00+08:00  阵雨   28       32        80     3级     东南风   10        1005
3  深圳  中国  2026-04-23T12:00:00+08:00  雷阵雨   29       33        85     4级     南风    8        1004
4  杭州  中国  2026-04-23T12:00:00+08:00  阴   22       24        65     2级     东北风   12        1009

📋 天气预报预览:
   城市         日期  白天天气  夜间天气  最高温度(℃)  最低温度(℃)  降水概率(%)  风力等级  风向  紫外线等级
0  北京  2026-04-24     晴     多云      28        18        15     2级     东风     中等
1  北京  2026-04-25     多云   小雨      26        17        45     3级     南风     弱
2  北京  2026-04-26     阴     晴      24        16        20     2级     北风     弱
3  北京  2026-04-27     晴     晴      30        19        0      1级     西风     强
4  北京  2026-04-28     多云   阴      27        18        30     2级     东风     中等

============================================================
🎉 数据获取完成!
📊 共获取 5 个城市的实时天气
📊 共获取 25 条预报数据
============================================================

📋 输出数据示例

实时天气数据表
城市天气温度(℃)湿度(%)风力等级风向
北京2545%3级北风
上海多云2360%2级东风
广州阵雨2880%3级东南风
深圳雷阵雨2985%4级南风
杭州2265%2级东北风
天气预报数据表
城市日期白天天气夜间天气最高温度最低温度降水概率
北京2026-04-24多云28℃18℃15%
北京2026-04-25多云小雨26℃17℃45%

🖼️ 实际运行效果截图描述

截图说明:控制台输出显示数据获取进度,数据成功保存为CSV和JSON文件。实时天气表格清晰展示各城市温度、湿度、风力等信息。

💡 常用免费天气API推荐

API名称提供商免费额度备注
心知天气心知天气400次/小时需要注册
和风天气和风天气1000次/天注册送额度
OpenWeatherMap国外60次/分钟国际通用
WeatherAPI国外100万次/月免费额度大

💡 知识点总结

技术点说明
requests.get()发送API请求
response.json()直接解析JSON响应
json.dumps()Python对象转JSON字符串
json.loads()JSON字符串转Python对象
pd.DataFrame()创建数据表格
datetime日期时间处理

❓ 常见问题与解决方案(Q&A)

Q1: 爬虫被网站封禁了怎么办? 😱

问题描述:请求被拒绝,返回403或验证码

解决方案

# 方案一:使用代理IP
proxies = {
    'http': 'http://user:password@proxy.com:8080',
    'https': 'http://user:password@proxy.com:8080',
}
response = requests.get(url, proxies=proxies)

# 方案二:更换User-Agent
user_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)}

# 方案三:降低请求频率
import time
time.sleep(random.uniform(3, 10))  # 3-10秒随机延时

# 方案四:使用Selenium/Playwright绕过
# 这两个工具可以模拟真实浏览器行为

Q2: 如何处理JavaScript渲染的页面? 🤔

问题描述:页面源码看不到数据,requests只能拿到空壳

解决方案

方案适用场景难度推荐指数
Selenium需要交互的页面⭐⭐⭐⭐⭐⭐
Playwright需要高性能⭐⭐⭐⭐⭐⭐⭐
requests-html轻量级JS支持⭐⭐⭐⭐⭐
API抓取有移动端/API⭐⭐⭐⭐⭐
# 推荐使用Playwright(速度快,更现代)
from playwright.async_api import async_playwright

async def scrape_js_page(url):
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        await page.goto(url)
        # 等待内容加载
        await page.wait_for_selector('your-selector')
        content = await page.content()
        await browser.close()
        return content

Q3: 数据解析出现乱码怎么办? 😤

问题描述:中文显示为乱码 锟斤拷\u8d5b

解决方案

# 方案一:设置正确的编码
response = requests.get(url)
response.encoding = 'utf-8'  # 或 'gbk', 'gb2312'

# 方案二:自动检测编码
import chardet
raw_data = response.content
result = chardet.detect(raw_data)
response = raw_data.decode(result['encoding'])

# 方案三:BeautifulSoup自动处理
soup = BeautifulSoup(response.text, 'lxml')  # 自动识别编码

Q4: 如何存储爬取的大量数据? 💾

问题描述:数据量太大,CSV打开很慢

解决方案

import pandas as pd

# 小数据量:CSV
df = pd.DataFrame(data)
df.to_csv('data.csv', index=False)

# 大数据量:分块存储或使用数据库
# 方案一:分块CSV
for chunk in pd.read_csv('big_data.csv', chunksize=10000):
    process(chunk)

# 方案二:JSON Lines格式(每行一个JSON)
with open('data.jsonl', 'w', encoding='utf-8') as f:
    for item in data:
        f.write(json.dumps(item, ensure_ascii=False) + '\n')

# 方案三:SQLite(轻量级数据库)
import sqlite3
conn = sqlite3.connect('data.db')
df.to_sql('table_name', conn, if_exists='replace')

# 方案四:MongoDB(适合非结构化数据)
# from pymongo import MongoDB
# client = MongoDB('mongodb://localhost:27017/')

Q5: 如何定时执行爬虫任务? ⏰

问题描述:需要定期爬取数据,不想手动运行

解决方案

# 方案一:schedule库(简单定时)
import schedule
import time

def job():
    print("正在执行爬虫...")
    # 调用你的爬虫函数
    my_spider.run()

# 每小时执行一次
schedule.every(1).hours.do(job)

# 每天上午10点执行
schedule.every().day.at("10:00").do(job)

# 持续运行
while True:
    schedule.run_pending()
    time.sleep(1)

# 方案二:APScheduler(高级定时)
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('interval', hours=1)
def timed_job():
    print('每小时的第1分钟执行')
    
@scheduler.scheduled_job('cron', hour=10, minute=0)
def morning_job():
    print('每天上午10点执行')

scheduler.start()

# 方案三:Windows任务计划程序 / Linux cron
# 在命令行中使用:
# crontab -e
# 0 * * * * python /path/to/your/scraper.py

Q6: 如何绕过反爬虫机制? 🛡️

问题描述:网站有严格的反爬措施

解决方案

# 常用反反爬策略
class AntiBanSpider:
    def __init__(self):
        self.session = requests.Session()  # 使用Session保持会话
        
    def get_random_headers(self):
        """随机生成请求头"""
        user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64)...',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)...',
            'Mozilla/5.0 (X11; Linux x86_64)...',
        ]
        return {
            'User-Agent': random.choice(user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Referer': 'https://www.google.com/',  # 模拟从Google跳转
        }
    
    def get_with_proxy(self, url):
        """使用代理"""
        proxies = {
            'http': 'http://proxy:port',
            'https': 'https://proxy:port',
        }
        return self.session.get(url, headers=self.get_random_headers(), proxies=proxies)
    
    def smart_sleep(self):
        """智能延时"""
        time.sleep(random.uniform(3, 8))

Q7: 爬虫程序崩溃了怎么办? 💥

问题描述:长时间运行崩溃,数据丢失

解决方案

import logging
from functools import wraps

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='spider.log',
    filemode='a',
)

def retry_on_failure(max_retries=3, delay=5):
    """失败自动重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logging.error(f"第{i+1}次失败:{e}")
                    if i < max_retries - 1:
                        time.sleep(delay)
                    else:
                        raise
            return None
        return wrapper
    return decorator

# 使用示例
@retry_on_failure(max_retries=3, delay=10)
def fetch_data():
    # 可能失败的操作
    pass

# 添加超时控制
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("请求超时!")

signal.signal(signal.SIGALRM, timeout_handler)

def fetch_with_timeout(url, timeout=30):
    signal.alarm(timeout)
    try:
        response = requests.get(url)
        return response
    finally:
        signal.alarm(0)  # 重置闹钟

🎯 总结与延伸

📚 本文知识点回顾

案例技术栈适用场景难度
案例一requests + BeautifulSoup静态网页⭐⭐
案例二Selenium/Playwright动态网页⭐⭐⭐
案例三requests + JSON解析API接口

🚀 进阶学习方向

1. 分布式爬虫
# 使用Scrapy框架构建分布式爬虫
# scrapy startproject myproject
# scrapy genspider example example.com
2. 数据库存储
# MongoDB存储
from pymongo import MongoClient
client = MongoClient()
db = client['spider_db']
collection = db['products']
collection.insert_many(data)

# Redis缓存
import redis
r = redis.Redis()
r.set('key', 'value')
3. 异步爬虫
# aiohttp异步爬虫
import aiohttp
import asyncio

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)
4. 反爬技术进阶
  • IP代理池
  • Cookie池
  • 验证码识别(OCR)
  • 模拟登录(OAuth)

📖 推荐学习资源

资源类型推荐内容
书籍《Python网络数据采集》
文档Scrapy官方文档、Selenium官方文档
网站博客园、知乎、CSDN
GitHub搜索"awesome-python-scraper"

⚠️ 爬虫伦理与法律提醒

🔔 重要提醒

  1. 遵守网站的robots.txt协议
  2. 不要爬取个人隐私数据
  3. 不要用于商业非法用途
  4. 合理控制请求频率
  5. 尊重数据版权

🎉 结语

通过本文的三个实战案例,相信你已经掌握了Python爬虫的核心技能:

✅ 静态网页爬虫:requests + BeautifulSoup
✅ 动态网页爬虫:Selenium + Playwright  
✅ API数据抓取:requests + JSON解析

记住:技术本身是中立的,关键在于如何使用。做一个有道德、有底线的爬虫工程师!


📌 标签

#Python #爬虫 #数据分析 #自动化 #WebScraping #Requests #BeautifulSoup #Selenium #Playwright #数据采集 #网络爬虫 #Python教程


📌 作者:AI效率玩家
📌 首发平台:CSDN博客
📌 原文链接:https://blog.csdn.net/…
📌 版权声明:本文为作者原创,转载须注明出处
📌 日期:2026-04-23


如果本文对你有帮助,欢迎点赞、评论、收藏! 🙏

如有疑问,欢迎在评论区留言,我会第一时间回复!

更多推荐