Python微博爬虫架构设计：模块化、反爬策略与工程化实践

简介：本资源是一款面向Python中级开发者与数据采集研究者的微博爬虫实战项目，聚焦SinaWeibo平台用户画像、社交关系链及超级话题生态的数据抓取需求。项目含41个文件，总大小10.99MB，涵盖12个核心.py源码（如weibo_cn_async.py、login.py、redis_cookies.py）、16个.pyc字节码、2个JPG/PNG图片（含SpiderFramework.jpg架构图）、1个YAML账号配置、1个DLL验证码识别库（yundamaAPI-x64.dll）、1个可执行文件（geckodriver.exe）及日志、配置、容器ID等辅助文件，结构完整，支持异步请求、Redis会话管理、验证码识别与反反爬策略。目前已有354人学习下载，适合需快速部署微博数据采集系统的开发者——既可直接运行获取用户/关注/粉丝/超话多维数据，又能通过清晰分层的src目录与__pycache__编译痕迹理解工程化爬虫的模块设计逻辑与调试路径。

1. 项目缘起：为什么我们需要一个“不一样”的微博爬虫？

做数据分析和舆情监控的朋友，对微博数据的需求是刚性的。市面上现成的爬虫工具和库不少，但用起来总感觉差点意思：要么是API调用限制太死，拿不到足够的历史数据；要么是模拟登录和反爬策略过于脆弱，跑几天就失效；再或者就是代码结构混乱，想加个数据清洗或者存到不同数据库的功能，得把源码翻个底朝天，改得心惊胆战。我自己在几个舆情分析项目里就深受其苦，所以决定动手写一个结构清晰、易于扩展、且能应对常见反爬机制的微博爬虫——这就是SinaWeiboSpider的由来。

这个爬虫的核心目标很明确：稳定、高效、可维护地获取微博的公开数据。它不是一个追求极限速度的“秒杀”式爬虫，而是一个强调工程化、适合集成到实际生产数据分析流水线中的工具。我会基于Python来实现，因为Python在数据处理和快速原型开发上的优势无可比拟。接下来，我会从设计思路、核心模块拆解、关键代码实现以及最重要的——实战中的避坑经验，来完整分享这个爬虫的设计源码。无论你是想学习爬虫架构，还是急需一个能直接上手的微博数据采集方案，这篇文章都能给你提供一条清晰的路径。

2. 整体架构设计：模块化与可扩展性优先

在设计之初，我就摒弃了把所有功能塞进一个脚本的“面条式”代码。一个健壮的爬虫应该像乐高积木，每个模块职责单一，通过清晰的接口拼装在一起。这样不仅便于调试和维护，未来要更换登录方式、解析规则或存储引擎时，也只需要动其中一个模块，不会牵一发而动全身。

2.1 核心模块划分

整个SinaWeiboSpider项目我分成了以下几个核心目录和模块：

sina_weibo_spider/ ├── core/ # 核心引擎 │ ├── __init__.py │ ├── downloader.py # 下载器，负责HTTP请求 │ ├── parser.py # 解析器，负责从HTML/JSON中提取数据 │ └── scheduler.py # 调度器，管理请求队列和去重 ├── spiders/ # 爬虫逻辑定义 │ ├── __init__.py │ └── weibo_spider.py # 具体的微博爬虫，定义爬取规则 ├── items/ # 数据模型定义 │ ├── __init__.py │ └── weibo_item.py # 微博数据项，定义字段结构 ├── pipelines/ # 数据处理管道 │ ├── __init__.py │ ├── validation_pipeline.py # 数据清洗与验证 │ └── storage_pipeline.py # 数据存储（文件、数据库） ├── middlewares/ # 中间件 │ ├── __init__.py │ ├── user_agent_middleware.py # UA轮换 │ └── proxy_middleware.py # 代理IP管理 ├── utils/ # 工具函数 │ ├── __init__.py │ ├── logger.py # 日志配置 │ ├── encrypt.py # 加密解密辅助（如密码） │ └── tools.py # 通用工具（时间转换、字符串处理） ├── config/ # 配置文件 │ └── settings.py ├── requirements.txt └── run.py # 项目启动入口

为什么这么设计？

• core/: 这是爬虫的“发动机”。下载器、解析器、调度器三者分离，符合经典爬虫架构。比如，当微博前端改版，只需要修改parser.py中的解析规则，下载逻辑完全不用动。
• spiders/: 存放具体的爬虫类。这里定义了“爬什么”和“怎么爬”的业务逻辑。如果你后续想爬微博热搜榜或者用户信息，可以在这里新建一个hotsearch_spider.py，复用core里的引擎。
• items/: 使用dataclass或attrs库定义数据模型。这比用字典更规范，能提前定义好每个字段的类型，在后续数据清洗和入库时能避免很多低级错误，比如时间戳格式混乱。
• pipelines/: 数据处理的流水线。爬取到的原始数据往往很“脏”，在这里进行清洗、去重、验证，然后再决定是存入CSV、MySQL还是MongoDB。管道可以配置多个，按顺序执行。
• middlewares/: 用于处理请求和响应的“钩子”。反爬的核心战场就在这里。通过中间件，我们可以无侵入地为每个请求动态添加代理、更换User-Agent，或者对响应进行预处理。
• config/settings.py: 所有可配置的参数集中管理，如并发数、下载延迟、重试次数、数据库连接字符串等。这样调整参数时不用去代码里大海捞针。

2.2 数据流与控制流

整个爬虫的运行流程可以概括为：

1. run.py启动，加载配置，初始化各模块。
2. scheduler从weibo_spider获取初始请求（如某个用户的微博列表页URL）。
3. scheduler将请求交给downloader。
4. downloader在发出请求前，会依次经过middlewares（如添加代理UA）。
5. 获取响应后，downloader将响应交给weibo_spider中指定的回调函数（通常在parser中）。
6. parser解析响应，生成两种结果：一是提取出的WeiboItem数据对象，二是新的需要继续爬取的Request对象（如下一页链接）。
7. 新的Request会回到scheduler排队，WeiboItem则被送入pipelines进行处理。
8. pipelines依次对数据进行清洗和存储。
9. 循环执行3-8步，直到scheduler中的请求队列为空或达到停止条件。

这个流程清晰地将“抓取”、“解析”、“存储”解耦，每个环节都可以独立优化和替换。

3. 核心模块源码实现与关键技术点

接下来，我们深入到几个最关键模块的代码层面，看看具体是怎么实现的，以及为什么这么实现。

3.1 配置管理 (config/settings.py)

我把所有配置项集中在这里，并使用Python的dataclass来管理，这样可以利用类型提示，避免配置项拼写错误。

from dataclasses import dataclass from typing import List, Optional @dataclass class Settings: # 网络请求相关 CONCURRENT_REQUESTS: int = 4 # 并发请求数，太高容易被封 DOWNLOAD_DELAY: float = 1.5 # 下载延迟秒数，模拟人类操作 RETRY_TIMES: int = 3 # 失败重试次数 TIMEOUT: int = 10 # 请求超时时间 DEFAULT_REQUEST_HEADERS: dict = None # 默认请求头 # 反爬相关 USER_AGENTS: List[str] = None # User-Agent列表，用于轮换 PROXY_LIST: List[str] = None # 代理IP列表，格式如 `http://ip:port` USE_PROXY: bool = False # 是否启用代理 # 目标数据相关 KEYWORDS: List[str] = None # 搜索关键词 TARGET_USER_IDS: List[str] = None # 目标用户ID START_DATE: str = None # 爬取起始日期 END_DATE: str = None # 爬取结束日期 # 存储相关 SAVE_TO_CSV: bool = True CSV_FILE_PATH: str = './data/weibo_data.csv' SAVE_TO_MYSQL: bool = False MYSQL_CONFIG: dict = None # 日志相关 LOG_LEVEL: str = 'INFO' LOG_FILE: str = './logs/spider.log' def __post_init__(self): # 初始化默认值，避免可变对象的共享问题 if self.DEFAULT_REQUEST_HEADERS is None: self.DEFAULT_REQUEST_HEADERS = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', } if self.USER_AGENTS is None: # 这里应该从一个外部文件或列表中读取，这里简写 self.USER_AGENTS = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...', # ... 更多UA ] if self.MYSQL_CONFIG is None: self.MYSQL_CONFIG = { 'host': 'localhost', 'port': 3306, 'user': 'root', 'password': '', 'database': 'weibo', 'charset': 'utf8mb4' # 必须utf8mb4，支持存储Emoji }

关键点解析：

• 使用dataclass：这比传统的字典配置更安全、更直观。你可以用settings = Settings()实例化，然后用settings.CONCURRENT_REQUESTS访问，IDE会有自动补全和类型检查。
• __post_init__方法：用于初始化那些复杂的默认值。注意，像USER_AGENTS这样的列表，如果在类属性中直接定义[]，所有实例会共享同一个列表对象，可能导致意外修改。放在__post_init__里初始化更安全。
• utf8mb4字符集：这是存储微博数据（尤其是包含Emoji、特殊符号的文本）时必须注意的。MySQL的utf8并非真正的UTF-8，最大只支持3字节，存不了4字节的Emoji，会导致插入失败。utf8mb4才是完整的UTF-8支持。

3.2 数据模型定义 (items/weibo_item.py)

清晰的数据模型是保证数据质量的基石。我使用pydantic库，因为它不仅提供了数据验证，还能轻松处理JSON序列化/反序列化。

from pydantic import BaseModel, Field, validator from datetime import datetime from typing import Optional, List class WeiboItem(BaseModel): """微博数据项模型""" weibo_id: str = Field(..., description="微博唯一ID") user_id: str = Field(..., description="发布者用户ID") screen_name: str = Field(..., description="发布者昵称") text: str = Field(..., description="微博正文") article_url: Optional[str] = Field(None, description="头条文章链接") publish_time: datetime = Field(..., description="发布时间") publish_tool: Optional[str] = Field(None, description="发布设备") reposts_count: int = Field(default=0, description="转发数") comments_count: int = Field(default=0, description="评论数") attitudes_count: int = Field(default=0, description="点赞数") pics: Optional[List[str]] = Field(default=None, description="图片链接列表") video_url: Optional[str] = Field(None, description="视频链接") location: Optional[str] = Field(None, description="定位信息") topics: Optional[List[str]] = Field(default=None, description="话题列表") at_users: Optional[List[str]] = Field(default=None, description="@的用户列表") crawl_time: datetime = Field(default_factory=datetime.now, description="爬取时间") @validator('publish_time', pre=True) def parse_publish_time(cls, v): """将微博页面中的时间字符串转换为datetime对象""" if isinstance(v, datetime): return v if isinstance(v, str): # 微博时间格式多样，如“刚刚”、“2分钟前”、“今天 10:20”、“03-15”、“2022-03-15” # 这里需要一个复杂的解析函数，篇幅所限，仅示意 # 实际项目中，我会写一个专门的 time_parser 函数 from utils.tools import parse_weibo_time return parse_weibo_time(v) raise ValueError(f'无法解析的时间格式: {type(v)}') class Config: json_encoders = { datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S') }

关键点解析：

• pydantic的优势：它会在实例化时自动进行类型验证。如果你尝试给reposts_count赋一个字符串，它会立刻报错，而不是把脏数据带到后面的流程。Field类提供了丰富的元数据，如描述、默认值。
• validator装饰器：这是pydantic的杀手级功能。微博页面的发布时间格式千奇百怪（“刚刚”、“昨天”、“3月15日”），我们可以在数据进入模型的第一时间，通过validator将其统一转换为标准的datetime对象，后续处理无比方便。
• default_factory：crawl_time使用default_factory=datetime.now，这意味着每次创建WeiboItem实例时，都会自动生成当前时间作为爬取时间，无需手动传入。
• json_encoders：定义了模型在转换为JSON字符串时，如何序列化datetime对象。这保证了存储到JSON文件或NoSQL数据库时的格式一致性。

3.3 下载器与反爬中间件 (core/downloader.py & middlewares/)

下载器是直接与微博服务器对话的模块，也是反爬攻防战的前线。一个健壮的下载器必须包含重试、超时、异常处理和中间件扩展能力。

# core/downloader.py 核心部分 import aiohttp import asyncio from typing import Optional from utils.logger import get_logger logger = get_logger(__name__) class AsyncDownloader: """基于aiohttp的异步下载器""" def __init__(self, settings, middlewares=None): self.settings = settings self.middlewares = middlewares or [] self.session: Optional[aiohttp.ClientSession] = None self.semaphore = asyncio.Semaphore(settings.CONCURRENT_REQUESTS) async def __aenter__(self): # 创建aiohttp会话，统一管理连接池和cookies timeout = aiohttp.ClientTimeout(total=self.settings.TIMEOUT) self.session = aiohttp.ClientSession(timeout=timeout, headers=self.settings.DEFAULT_REQUEST_HEADERS) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() async def fetch(self, request): """执行单个请求""" url = request.url method = request.method kwargs = request.kwargs # 可能包含data, json, params等 # 1. 请求预处理：经过所有中间件 for middleware in self.middlewares: request = await middleware.process_request(request) if request is None: # 中间件可能中断请求 return None async with self.semaphore: # 控制并发 await asyncio.sleep(self.settings.DOWNLOAD_DELAY) # 延迟 for attempt in range(self.settings.RETRY_TIMES + 1): try: async with self.session.request(method, url, **kwargs) as response: response.raise_for_status() # 检查HTTP状态码 html = await response.text() # 2. 响应后处理：经过所有中间件 for middleware in self.middlewares: html = await middleware.process_response(response, html, request) if html is None: return None return html except (aiohttp.ClientError, asyncio.TimeoutError) as e: logger.warning(f"请求失败 ({attempt+1}/{self.settings.RETRY_TIMES+1}): {url}, 错误: {e}") if attempt == self.settings.RETRY_TIMES: logger.error(f"请求最终失败: {url}") raise await asyncio.sleep(2 ** attempt) # 指数退避重试

关键点解析：

• 异步IO (aiohttp)：对于IO密集型的网络爬虫，异步能极大提升效率。使用async/await语法和aiohttp库，可以在单个线程内并发处理数十上百个请求。
• 信号量 (asyncio.Semaphore)：用于精确控制并发请求数，避免瞬间发起过多请求导致IP被封锁。
• 指数退避重试：await asyncio.sleep(2 ** attempt)。第一次重试等2秒，第二次等4秒，以此类推。这是一种礼貌且有效的重试策略，避免在服务器临时故障时持续轰炸。
• 中间件钩子：process_request和process_response是中间件的标准接口。这构成了一个强大的插件系统。

# middlewares/user_agent_middleware.py import random from core.request import Request class UserAgentMiddleware: """随机User-Agent中间件""" def __init__(self, user_agents): self.user_agents = user_agents async def process_request(self, request: Request): if self.user_agents: ua = random.choice(self.user_agents) if 'headers' not in request.kwargs: request.kwargs['headers'] = {} request.kwargs['headers']['User-Agent'] = ua return request

# middlewares/proxy_middleware.py import random from core.request import Request class ProxyMiddleware: """随机代理IP中间件""" def __init__(self, proxy_list, enabled=False): self.proxy_list = proxy_list or [] self.enabled = enabled async def process_request(self, request: Request): if self.enabled and self.proxy_list: proxy = random.choice(self.proxy_list) if 'proxy' not in request.kwargs: request.kwargs['proxy'] = proxy elif request.kwargs.get('proxy') is None: # 如果请求本身指定了proxy为None（即明确不用代理），则跳过 pass else: request.kwargs['proxy'] = proxy return request

实战心得：

• User-Agent池：不要只用一两个UA。最好准备几十个不同浏览器、不同版本的UA，并定期更新。可以从一些开源项目或网站上获取列表。
• 代理IP的质量：免费代理IP大多不稳定且速度慢。对于严肃的数据采集项目，建议使用付费的优质代理服务，并按需购买动态住宅IP或数据中心IP。在中间件里可以加入代理IP的健康检查机制，自动剔除失效的IP。
• Cookie管理：对于需要登录才能访问的数据，aiohttp.ClientSession会自动管理Cookie。但更复杂的场景（如Cookie过期自动重新登录），可能需要一个专门的CookieMiddleware来维护。

3.4 解析器：应对动态渲染与数据接口 (core/parser.py & spiders/weibo_spider.py)

微博的数据获取主要有两种方式：解析HTML页面和调用内部JSON API。前者直观但易受页面改版影响；后者稳定但需要分析网络请求。

方式一：解析HTML页面（以用户主页为例）早期爬虫多采用此方式，但如今微博页面JavaScript渲染越来越多，直接拿到的HTML可能不包含完整数据。

# 在 spiders/weibo_spider.py 中 from bs4 import BeautifulSoup import re from items.weibo_item import WeiboItem class WeiboSpider: def parse_user_tweets(self, html: str): """解析用户主页微博列表HTML""" soup = BeautifulSoup(html, 'lxml') tweet_cards = soup.find_all('div', class_=re.compile('WB_cardwrap.*')) # 类名可能变化 items = [] for card in tweet_cards: try: # 提取微博ID - 通常藏在某个属性里 mid = card.get('mid') or card.find('div', attrs={'mid': True}) if not mid: continue weibo_id = mid.get('mid') if hasattr(mid, 'get') else mid # 提取正文 - 注意处理长微博和转发 text_node = card.find('div', class_=re.compile('WB_text.*')) text = text_node.get_text(strip=True) if text_node else '' # 提取发布时间 - 这是一个难点，时间格式不统一 time_node = card.find('a', class_=re.compile('S_txt2.*')) publish_time_str = time_node.get('title') or time_node.get_text(strip=True) # 构造数据项 item = WeiboItem( weibo_id=weibo_id, user_id=self.current_user_id, screen_name=self.current_screen_name, text=text, publish_time=publish_time_str, # 这里会触发pydantic的validator进行转换 # ... 提取其他字段 ) items.append(item) except Exception as e: self.logger.error(f"解析微博卡片失败: {e}, 卡片内容: {card[:200]}") continue return items

方式二：调用内部API（推荐）通过浏览器开发者工具的“网络”(Network)选项卡，观察微博页面加载时发出的XHR/Fetch请求，往往能找到返回结构化JSON数据的接口。这种方式更稳定，数据也更干净。

# 在 spiders/weibo_spider.py 中增加API解析方法 import json class WeiboSpider: def parse_timeline_api(self, json_data: dict): """解析微博时间线API返回的JSON数据""" # 微博API返回的数据结构通常很嵌套 statuses = json_data.get('data', {}).get('statuses', []) items = [] for status in statuses: try: # 直接映射JSON字段到我们的数据模型 item = WeiboItem( weibo_id=str(status['id']), user_id=str(status['user']['id']), screen_name=status['user']['screen_name'], text=status['text_raw'], # 注意：用text_raw而非text，后者包含HTML标签 publish_time=status['created_at'], # API返回标准时间格式 reposts_count=status.get('reposts_count', 0), comments_count=status.get('comments_count', 0), attitudes_count=status.get('attitudes_count', 0), pics=[pic['large']['url'] for pic in status.get('pic_infos', {}).values()] if status.get('pic_infos') else None, # ... 其他字段 ) items.append(item) except KeyError as e: self.logger.warning(f"JSON字段缺失 {e}，跳过该条微博。数据: {status.get('id')}") continue return items

关键点解析与避坑指南：

• text_rawvstext：这是最容易踩的坑！微博API返回的text字段是经过HTML转义的字符串，里面包含了<br>换行符、<a>链接标签等。而text_raw才是纯净的原文。如果你把text直接当作文本分析，会引入大量HTML噪音。
• 时间处理：API返回的created_at通常是“Mon Mar 20 10:00:00 +0800 2023”这种格式，Python可以用datetime.strptime(created_at, '%a %b %d %H:%M:%S %z %Y')来解析。务必注意时区信息。
• 反爬机制：微博的API请求通常需要携带Cookie（尤其是登录后的SUB和SUBP令牌）以及一个动态生成的XSRF-TOKEN（可能在Cookie或请求头中）。你需要通过模拟登录或手动从浏览器复制Cookie来获取这些凭证。重要提示：这些令牌有有效期，需要设计刷新机制。
• BeautifulSoup的备用性：尽管推荐用API，但保留HTML解析代码作为备用方案是明智的。当API接口发生变化或无法访问时，可以快速回退到HTML解析。解析时，尽量使用re.compile进行模糊匹配，因为微博的CSS类名经常微调。

3.5 数据管道：清洗、验证与存储 (pipelines/)

原始数据爬下来后，必须经过清洗才能使用。管道(Pipeline)模式让这个过程井然有序。

# pipelines/validation_pipeline.py import re from items.weibo_item import WeiboItem class ValidationPipeline: """数据清洗与验证管道""" def process_item(self, item: WeiboItem): # 1. 清理文本：去除多余空白、不可见字符 if item.text: item.text = self._clean_text(item.text) # 2. 提取话题和@用户 if item.text: item.topics = self._extract_topics(item.text) item.at_users = self._extract_at_users(item.text) # 3. 验证必要字段 if not item.weibo_id or not item.user_id: raise DropItem(f"缺失必要字段: weibo_id或user_id为空") # 4. 去重检查 (简单示例，实际可能基于数据库) if self._is_duplicate(item.weibo_id): raise DropItem(f"重复微博: {item.weibo_id}") return item def _clean_text(self, text): """清理文本，移除HTML标签、多余换行等""" # 移除HTML标签 text = re.sub(r'<[^>]+>', '', text) # 将多个连续空白字符（包括换行）替换为单个空格 text = re.sub(r'\s+', ' ', text) return text.strip() def _extract_topics(self, text): """从文本中提取话题，如 #这是一个话题# """ topics = re.findall(r'#([^#]+?)#', text) return topics if topics else None def _extract_at_users(self, text): """从文本中提取@的用户，如 @用户名 """ at_users = re.findall(r'@([\w\u4e00-\u9fa5\-]+)', text) return at_users if at_users else None def _is_duplicate(self, weibo_id): # 这里可以连接一个内存中的集合或Redis进行去重检查 # 简单示例：使用类属性存储已见ID（仅单进程有效） if not hasattr(self, '_seen_ids'): self._seen_ids = set() if weibo_id in self._seen_ids: return True self._seen_ids.add(weibo_id) return False

# pipelines/storage_pipeline.py import csv import pymysql from pymysql import MySQLError from items.weibo_item import WeiboItem from config.settings import Settings class StoragePipeline: """数据存储管道（支持CSV和MySQL）""" def __init__(self, settings: Settings): self.settings = settings self.csv_file = None self.csv_writer = None self.mysql_conn = None def open_spider(self): """爬虫启动时调用""" # 初始化CSV写入器 if self.settings.SAVE_TO_CSV: import os os.makedirs(os.path.dirname(self.settings.CSV_FILE_PATH), exist_ok=True) self.csv_file = open(self.settings.CSV_FILE_PATH, 'a', newline='', encoding='utf-8-sig') # 注意编码 fieldnames = list(WeiboItem.schema()['properties'].keys()) # 从pydantic模型获取字段名 self.csv_writer = csv.DictWriter(self.csv_file, fieldnames=fieldnames) if self.csv_file.tell() == 0: # 文件为空时写入表头 self.csv_writer.writeheader() # 初始化MySQL连接 if self.settings.SAVE_TO_MYSQL: try: self.mysql_conn = pymysql.connect(**self.settings.MYSQL_CONFIG) self._create_table_if_not_exists() except MySQLError as e: self.logger.error(f"连接MySQL失败: {e}") self.mysql_conn = None def process_item(self, item: WeiboItem): item_dict = item.dict() # 将pydantic模型转为字典 # 存储到CSV if self.settings.SAVE_TO_CSV and self.csv_writer: self.csv_writer.writerow(item_dict) # 存储到MySQL if self.settings.SAVE_TO_MYSQL and self.mysql_conn: self._save_to_mysql(item_dict) return item def _create_table_if_not_exists(self): """创建微博数据表（如果不存在）""" create_table_sql = """ CREATE TABLE IF NOT EXISTS weibo ( id INT AUTO_INCREMENT PRIMARY KEY, weibo_id VARCHAR(50) NOT NULL UNIQUE COMMENT '微博唯一ID', user_id VARCHAR(30) NOT NULL COMMENT '用户ID', screen_name VARCHAR(100) COMMENT '用户昵称', text TEXT COMMENT '微博正文', article_url VARCHAR(500) COMMENT '文章链接', publish_time DATETIME COMMENT '发布时间', publish_tool VARCHAR(100) COMMENT '发布工具', reposts_count INT DEFAULT 0 COMMENT '转发数', comments_count INT DEFAULT 0 COMMENT '评论数', attitudes_count INT DEFAULT 0 COMMENT '点赞数', pics JSON COMMENT '图片链接数组', video_url VARCHAR(500) COMMENT '视频链接', location VARCHAR(200) COMMENT '定位信息', topics JSON COMMENT '话题数组', at_users JSON COMMENT '@用户数组', crawl_time DATETIME COMMENT '爬取时间', INDEX idx_user_id (user_id), INDEX idx_publish_time (publish_time), INDEX idx_weibo_id (weibo_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; """ with self.mysql_conn.cursor() as cursor: cursor.execute(create_table_sql) self.mysql_conn.commit() def _save_to_mysql(self, item_dict): """插入数据到MySQL""" # 准备SQL和参数 placeholders = ', '.join(['%s'] * len(item_dict)) columns = ', '.join(item_dict.keys()) sql = f"INSERT INTO weibo ({columns}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE crawl_time = VALUES(crawl_time)" # ON DUPLICATE KEY UPDATE 避免重复插入，仅更新爬取时间 try: with self.mysql_conn.cursor() as cursor: # 需要将字典值转换为元组，并处理JSON字段 values = [] for v in item_dict.values(): if isinstance(v, (list, dict)): import json v = json.dumps(v, ensure_ascii=False) values.append(v) cursor.execute(sql, tuple(values)) self.mysql_conn.commit() except MySQLError as e: self.logger.error(f"插入MySQL失败: {e}, SQL: {sql}, 数据: {item_dict.get('weibo_id')}") self.mysql_conn.rollback() def close_spider(self): """爬虫关闭时调用""" if self.csv_file: self.csv_file.close() if self.mysql_conn: self.mysql_conn.close()

关键点解析与避坑指南：

• CSV文件编码：使用utf-8-sig编码而非utf-8。utf-8-sig会在文件开头写入一个BOM（字节顺序标记），这样用Excel打开中文CSV文件时不会出现乱码。这是一个非常实用的小技巧。
• MySQL字段类型：
  • text字段使用TEXT类型，足够存储长微博。
  • pics、topics等列表字段，使用JSON类型存储。MySQL 5.7+支持JSON类型，查询和操作都很方便。插入前用json.dumps()序列化。
  • 再次强调字符集：CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci。这是存储社交媒体数据的铁律。
• 唯一索引与去重：在MySQL表上为weibo_id创建UNIQUE约束，并使用INSERT ... ON DUPLICATE KEY UPDATE语句。这样即使爬虫因中断重启，重复爬取到相同微博，也不会插入重复数据，只会更新crawl_time等字段。这是实现增量爬取和避免数据冗余的关键。
• 连接管理：在open_spider和close_spider中管理文件句柄和数据库连接，确保资源被正确打开和关闭，避免资源泄漏。
• 错误处理：数据库操作必须用try...except包裹，记录错误日志并执行回滚(rollback)，保证单条数据失败不影响整体任务，也便于事后排查。

4. 实战部署与高级策略

4.1 如何运行这个爬虫？

一个清晰的入口脚本能让使用和调度变得简单。

# run.py import asyncio import sys from config.settings import Settings from core.downloader import AsyncDownloader from core.scheduler import Scheduler from spiders.weibo_spider import WeiboSpider from middlewares.user_agent_middleware import UserAgentMiddleware from middlewares.proxy_middleware import ProxyMiddleware from pipelines.validation_pipeline import ValidationPipeline from pipelines.storage_pipeline import StoragePipeline async def main(): # 1. 加载配置 settings = Settings( CONCURRENT_REQUESTS=2, # 初始建议调低，稳定后再提高 DOWNLOAD_DELAY=2.0, TARGET_USER_IDS=['1669879400'], # 示例：某个用户的ID SAVE_TO_CSV=True, SAVE_TO_MYSQL=False, # 根据需求开启 ) # 2. 初始化中间件 middlewares = [ UserAgentMiddleware(settings.USER_AGENTS), ProxyMiddleware(settings.PROXY_LIST, settings.USE_PROXY), ] # 3. 初始化管道 pipelines = [ ValidationPipeline(), StoragePipeline(settings), ] # 4. 初始化核心组件 spider = WeiboSpider(settings) scheduler = Scheduler() async with AsyncDownloader(settings, middlewares) as downloader: # 5. 启动爬虫逻辑 await spider.start(scheduler) # 6. 主循环：调度器取请求 -> 下载器下载 -> 爬虫解析 -> 管道处理 while not scheduler.idle(): request = scheduler.next_request() if request: html = await downloader.fetch(request) if html: new_items, new_requests = spider.parse(request, html) # 处理新数据 for item in new_items: for pipeline in pipelines: try: item = pipeline.process_item(item) except DropItem: break # 如果某个管道丢弃了该项，则不再传递 # 将新请求加入调度器 for new_req in new_requests: scheduler.add_request(new_req) # 可以在这里加入一些进度日志 print("爬取任务完成！") if __name__ == '__main__': # 处理异步事件循环，兼容不同环境 if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) asyncio.run(main())

4.2 应对高级反爬策略

微博的反爬在不断升级，除了UA和代理，还需要注意：

• 频率限制：DOWNLOAD_DELAY不要设得太小，并发数CONCURRENT_REQUESTS也要保守。观察服务器返回的HTTP状态码，如果频繁出现418、429或重定向到验证页面，说明触发了频率限制，需要进一步降低速度或更换IP。
• JavaScript挑战：微博的部分页面（如登录页、某些动态内容）会使用JavaScript生成加密参数或进行人机验证。纯aiohttp无法执行JS。这时有两种选择：
  1. 逆向工程：使用浏览器开发者工具调试，找到JS生成关键参数的逻辑，然后用Python的execjs或PyExecJS库来执行这段JS代码。这需要一定的逆向能力。
  2. 无头浏览器：对于复杂的JS渲染和验证码，可以使用playwright或selenium控制无头浏览器（如Chrome）来模拟真人操作。这种方式资源消耗大、速度慢，但最接近真实用户。建议：将无头浏览器作为最后的手段，仅用于获取关键令牌（如登录后的Cookie），然后用这个Cookie去调用API接口。
• Cookie池与账号池：对于大规模爬取，单一账号的Cookie很快会失效或被限制。需要维护一个账号池，实现自动登录、Cookie刷新和轮换调度。这涉及到验证码识别（如打码平台）和登录状态保持，是一个复杂的子系统。

4.3 监控、日志与错误恢复

一个用于生产的爬虫必须有完善的监控和自愈能力。

• 结构化日志：使用Python的logging模块，为不同模块设置不同的日志级别（DEBUG, INFO, WARNING, ERROR）。将日志同时输出到控制台和文件，便于调试和后期审计。
• 关键指标监控：在代码中埋点，记录爬取速度（条/分钟）、请求成功率、重复率、各字段缺失率等。这些指标可以帮助你判断爬虫的健康状态和反爬强度。
• 断点续爬：调度器Scheduler应将待爬队列（pending）和已爬集合（seen）持久化（例如存到Redis或磁盘文件）。这样即使爬虫进程崩溃，重启后也能从上次中断的地方继续，而不是从头开始。Scheduler的idle()和next_request()方法需要与持久化存储联动。
• 告警机制：当错误率超过阈值、或长时间没有爬取到新数据时，可以通过邮件、钉钉、企业微信机器人发送告警，通知开发者及时介入排查。

5. 总结与个人体会

写一个爬虫，从能跑到稳定、高效、易维护，中间隔着无数个坑。这个SinaWeiboSpider的设计，凝聚了我过去几年在数据采集项目中的大量经验教训。最重要的体会是：不要过分追求一次性爬取所有数据，而是优先保证爬虫的长期稳定运行。宁可速度慢一点，也要把错误处理、日志记录、状态持久化做扎实。

在具体实践中，有几点心得可以分享： 第一，数据模型先行。花时间用pydantic定义好清晰的数据模型，后续的数据清洗、验证、存储会省力很多，数据质量也有保障。 第二，配置驱动。所有可变的参数（延迟、并发、目标ID、存储路径）一定要放到配置文件里，绝对不要硬编码在代码中。这是项目能否复用的关键。 第三，敬畏反爬。把网络请求想象成一次对话，你的爬虫行为越像真人，活得就越久。随机的延迟、合理的并发、高质量的代理IP和定期更换的UA，这些细节比任何奇技淫巧都管用。 第四，重视监控。爬虫一旦部署，就不是“写完了”，而是“开始运行了”。必须有日志和监控告诉你它是否还活着，活得怎么样。一个黑盒爬虫是可怕的。

这个项目的源码结构是经过多个项目迭代后的结果，它可能不是性能最高的，但在可读性、可维护性和扩展性上做了很多权衡。你可以直接基于它进行二次开发，比如增加对微博评论、用户关系的爬取，或者将存储后端换成Elasticsearch、MongoDB。希望这个设计和代码能为你提供一个坚实的起点，而不仅仅是又一个“玩具级”的爬虫脚本。

本文还有配套的精品资源，点击获取