简介本资源是一款面向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 数据流与控制流整个爬虫的运行流程可以概括为run.py启动加载配置初始化各模块。scheduler从weibo_spider获取初始请求如某个用户的微博列表页URL。scheduler将请求交给downloader。downloader在发出请求前会依次经过middlewares如添加代理UA。获取响应后downloader将响应交给weibo_spider中指定的回调函数通常在parser中。parser解析响应生成两种结果一是提取出的WeiboItem数据对象二是新的需要继续爬取的Request对象如下一页链接。新的Request会回到scheduler排队WeiboItem则被送入pipelines进行处理。pipelines依次对数据进行清洗和存储。循环执行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/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.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(default0, description转发数) comments_count: int Field(default0, description评论数) attitudes_count: int Field(default0, description点赞数) pics: Optional[List[str]] Field(defaultNone, description图片链接列表) video_url: Optional[str] Field(None, description视频链接) location: Optional[str] Field(None, description定位信息) topics: Optional[List[str]] Field(defaultNone, description话题列表) at_users: Optional[List[str]] Field(defaultNone, description的用户列表) crawl_time: datetime Field(default_factorydatetime.now, description爬取时间) validator(publish_time, preTrue) 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_factorycrawl_time使用default_factorydatetime.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, middlewaresNone): 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(totalself.settings.TIMEOUT) self.session aiohttp.ClientSession(timeouttimeout, headersself.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请求失败 ({attempt1}/{self.settings.RETRY_TIMES1}): {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, enabledFalse): 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(stripTrue) 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(stripTrue) # 构造数据项 item WeiboItem( weibo_idweibo_id, user_idself.current_user_id, screen_nameself.current_screen_name, texttext, publish_timepublish_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_idstr(status[id]), user_idstr(status[user][id]), screen_namestatus[user][screen_name], textstatus[text_raw], # 注意用text_raw而非text后者包含HTML标签 publish_timestatus[created_at], # API返回标准时间格式 reposts_countstatus.get(reposts_count, 0), comments_countstatus.get(comments_count, 0), attitudes_countstatus.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(fJSON字段缺失 {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_okTrue) self.csv_file open(self.settings.CSV_FILE_PATH, a, newline, encodingutf-8-sig) # 注意编码 fieldnames list(WeiboItem.schema()[properties].keys()) # 从pydantic模型获取字段名 self.csv_writer csv.DictWriter(self.csv_file, fieldnamesfieldnames) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_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 fINSERT 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_asciiFalse) 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()序列化。再次强调字符集CHARSETutf8mb4 COLLATEutf8mb4_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_REQUESTS2, # 初始建议调低稳定后再提高 DOWNLOAD_DELAY2.0, TARGET_USER_IDS[1669879400], # 示例某个用户的ID SAVE_TO_CSVTrue, SAVE_TO_MYSQLFalse, # 根据需求开启 ) # 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。这时有两种选择逆向工程使用浏览器开发者工具调试找到JS生成关键参数的逻辑然后用Python的execjs或PyExecJS库来执行这段JS代码。这需要一定的逆向能力。无头浏览器对于复杂的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。希望这个设计和代码能为你提供一个坚实的起点而不仅仅是又一个“玩具级”的爬虫脚本。本文还有配套的精品资源点击获取