简介本资源是一款面向Python开发者与数据采集初学者的微博爬虫实战项目聚焦SinaWeibo平台用户画像、社交关系链及超级话题生态的数据抓取需求有效解决社交媒体公开数据获取难、反爬适配弱、结构化存储缺等实际问题。压缩包共41个文件总大小10.99MB涵盖12个核心Python源码如weibo_cn_async.py、redis_cookies.py、code_recognize.py、16个pyc字节码、2个JPG/PNG图片含SpiderFramework.jpg架构图、1个YAML账号配置、1个DLL验证码识别库yundamaAPI-x64.dll、1个可执行文件geckodriver.exe及日志、配置、属性等辅助文件模块分工明确支持登录鉴权、Cookie持久化、验证码识别与Redis缓存集成。目前已有354人学习下载提供完整可运行框架、异步多进程混合抓取逻辑、容器ID动态提取机制及logging.conf日志规范开箱即用便于二次开发与教学演示。1. 项目缘起为什么我们需要一个“聪明”的微博爬虫做舆情分析、用户画像研究或者只是想追踪某个话题的实时动态微博都是一个绕不开的数据金矿。但当你真正动手去挖的时候很快就会发现直接用requests库去请求微博页面要么拿不到数据要么拿到的是一堆乱码或者干脆被反爬机制直接“抬走”。这就是为什么我们需要一个专门为微博设计的爬虫——SinaWeiboSpider。它不是一个简单的网页下载器而是一个需要理解微博数据接口、模拟用户行为、处理动态加载和应对反爬策略的复杂工具。我最初写这个爬虫是因为一个市场分析项目需要持续追踪几个竞品官微的动态和用户反馈。手动复制粘贴不现实通用的爬虫框架又对微博这种“重防御”的站点水土不服。于是我决定基于Python从零开始构建一个稳定、高效且易于维护的微博数据采集器。这个过程踩了无数的坑也总结了不少经验。今天我就把这个项目的设计思路和核心源码拆解开来分享给大家。无论你是想学习爬虫进阶技巧还是急需一个能跑起来的微博数据采集方案这篇文章都能给你提供一条清晰的路径。2. 核心挑战解析微博的反爬机制与应对策略在动手写代码之前我们必须先搞清楚对手是谁。微博的反爬机制经过多年迭代已经相当成熟主要围绕以下几个方面构建防御。我们的爬虫设计本质上就是与这些机制进行“友好”的周旋。2.1 动态数据加载与接口分析这是第一个也是最关键的门槛。微博的网页内容尤其是时间线、评论、转发列表几乎全部通过Ajax接口动态加载。你在浏览器F12的Network面板里看到的不再是简单的HTML而是一个个XHRFetch请求返回的是结构化的JSON数据。如何应对我们的爬虫必须放弃解析HTML转向直接模拟这些Ajax请求。这意味着你需要使用浏览器开发者工具在微博页面如用户主页打开F12切换到Network网络面板筛选XHR或Fetch请求。寻找数据接口滚动页面观察新出现的请求。通常获取微博列表的接口URL会包含“containerid”和“uid”等关键参数。分析请求参数重点查看请求的Headers特别是Cookie和Authorization如果有。Query String Parameters或Payload里的参数也需要仔细研究比如since_id,page等用于翻页的参数。解析响应数据接口返回的是JSON结构清晰。你需要找到包含微博正文、发布时间、转发/评论/点赞数、用户信息等字段的路径。注意微博的接口参数和格式可能会不定期变动。一个稳定的爬虫不能硬编码接口URL最好能设计一种机制在爬虫启动时或运行中自动探测或更新接口信息虽然实现复杂但这是高阶玩法。2.2 用户身份验证Cookie与Token绝大多数有价值的微博数据接口都需要用户登录状态。爬虫如何获得这个状态Cookie这是最传统也是目前最主要的方式。用户登录后浏览器会保存一串Cookie其中包含SUB、SUBP等关键字段用于标识会话。爬虫可以直接使用这些Cookie来模拟登录状态。获取方式手动登录后从浏览器开发者工具的Application或Storage标签页中复制Cookie字符串。或者通过Selenium等自动化工具控制浏览器完成登录流程再提取Cookie。风险Cookie会过期。需要定期更新。同一个Cookie高频访问会被异常检测。Token/Authorization在一些新的或官方的API中可能会采用Bearer Token等认证方式。这通常需要更复杂的OAuth流程对于普通爬虫来说成本较高。策略选择对于个人或小规模爬取手动获取并定期更新Cookie是最快最直接的方法。我们的示例爬虫也将基于此方案。你需要将Cookie字符串保存在配置文件或环境变量中并在请求头中携带。2.3 请求频率限制与IP封禁微博会对异常高频的请求进行限制轻则返回错误码重则封禁IP一段时间。频率控制在代码中必须在每次请求之间加入随机延时。例如使用time.sleep(random.uniform(1, 3))模拟人类浏览的间隔。不要用固定间隔那太像机器了。IP代理池对于大规模、长时间爬取使用代理IP池是必须的。免费的代理不稳定付费的代理服务是更靠谱的选择。爬虫需要能够从代理池中自动获取并切换IP。User-Agent轮换使用单一的User-Agent也容易被识别。可以准备一个列表每次请求随机选择一个。2.4 数据加密与参数签名部分关键接口如发布微博、私信等的请求参数可能被加密或需要签名。这对于只做公开数据采集的爬虫来说通常不是首要问题。但如果你发现某个接口返回的数据乱码或请求总是失败可能需要检查其是否启用了加密。逆向分析JavaScript加密逻辑是爬虫工程师的高阶技能这里不展开。3. 爬虫架构设计与模块划分一个健壮的爬虫不应该把所有代码都堆在一个文件里。良好的架构能让代码更清晰、更易维护、也更容易扩展。我将这个SinaWeiboSpider设计为以下几个核心模块SinaWeiboSpider/ ├── spider.py # 爬虫主调度器负责流程控制 ├── fetcher.py # 网络请求模块封装请求逻辑和反爬策略 ├── parser.py # 数据解析模块从JSON中提取结构化信息 ├── storage.py # 数据存储模块支持保存到文件或数据库 ├── config.py # 配置文件存放Cookie、代理、关键词等 └── utils.py # 工具函数如日志、延时、异常处理3.1 网络请求模块 (fetcher.py)这是爬虫的“手”和“脚”负责与微博服务器通信。它的核心是健壮性和伪装能力。# fetcher.py import requests import time import random from typing import Optional, Dict, Any import logging logger logging.getLogger(__name__) class WeiboFetcher: def __init__(self, cookies: str, use_proxy: bool False, proxy_pool: list None): 初始化Fetcher。 :param cookies: 从浏览器复制的Cookie字符串。 :param use_proxy: 是否启用代理。 :param proxy_pool: 代理IP池列表格式如 [http://ip1:port, http://ip2:port] self.session requests.Session() # 设置请求头模拟浏览器 self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Cookie: cookies, # 关键注入Cookie Referer: https://weibo.com/ } self.session.headers.update(self.headers) self.use_proxy use_proxy self.proxy_pool proxy_pool or [] self.current_proxy_index 0 # 请求间隔配置 self.min_delay 2 self.max_delay 5 def _get_proxy(self) - Optional[Dict]: 从代理池中获取一个代理配置。 if not self.use_proxy or not self.proxy_pool: return None proxy self.proxy_pool[self.current_proxy_index] self.current_proxy_index (self.current_proxy_index 1) % len(self.proxy_pool) logger.debug(f使用代理: {proxy}) return {http: proxy, https: proxy} def _random_delay(self): 执行随机延时模拟人类操作。 delay random.uniform(self.min_delay, self.max_delay) logger.debug(f等待 {delay:.2f} 秒...) time.sleep(delay) def fetch_json(self, url: str, params: Optional[Dict] None) - Optional[Dict[str, Any]]: 发送GET请求获取JSON数据。 :param url: 接口URL :param params: 查询参数 :return: 解析后的JSON字典失败返回None self._random_delay() # 每次请求前先等待 proxies self._get_proxy() try: response self.session.get(url, paramsparams, proxiesproxies, timeout15) response.raise_for_status() # 检查HTTP状态码是否为200 # 微博接口通常返回JSON但需要检查状态码 json_data response.json() if json_data.get(ok) 1: # 微博接口常见的成功标志 return json_data.get(data) else: logger.error(f接口返回错误: {json_data.get(msg)}) return None except requests.exceptions.RequestException as e: logger.error(f网络请求失败: {url}, 错误: {e}) # 这里可以添加重试逻辑 return None except ValueError as e: logger.error(fJSON解析失败: {url}, 响应内容: {response.text[:200]}) return None # 示例获取用户主页微博列表的接口调用 # 这个接口URL和参数需要你通过分析实际网络请求获得 # url https://weibo.com/ajax/statuses/mymblog # params { # uid: 1234567890, # 用户ID # page: 1, # feature: 0 # } # fetcher WeiboFetcher(cookiesyour_cookies) # data fetcher.fetch_json(url, params)关键点解析使用Sessionrequests.Session()可以自动保持Cookie避免每次请求都手动设置。完整的Headers除了Cookie和User-AgentReferer来源页有时也很重要模拟从微博页面跳转过来的请求。异常处理网络请求充满不确定性必须用try...except包裹并记录详细的错误日志便于排查。延时与代理_random_delay和_get_proxy是反爬的核心。即使不用代理延时也必不可少。3.2 数据解析模块 (parser.py)这个模块是爬虫的“大脑”负责把fetcher拿回来的原始JSON数据转换成我们需要的、结构清晰的Python对象如字典或自定义类。# parser.py import json from datetime import datetime from typing import List, Dict, Any import logging logger logging.getLogger(__name__) class WeiboParser: 微博数据解析器。 staticmethod def parse_weibo_list(data: Dict) - List[Dict]: 解析微博列表JSON数据。 :param data: fetch_json返回的‘data’字段对应的字典。 :return: 解析后的微博字典列表。 weibo_list [] if not data or list not in data: logger.warning(数据中未找到微博列表。) return weibo_list for item in data[list]: try: weibo { id: item.get(id), # 微博ID bid: item.get(bid), # 微博bid另一种ID形式 text: item.get(text_raw), # 原始正文推荐text可能包含HTML标签 created_at: item.get(created_at), # 创建时间字符串 timestamp: WeiboParser._parse_time(item.get(created_at)), # 转为时间戳 source: item.get(source, ).replace(来自, ), # 发布来源如“iPhone客户端” user: { id: item.get(user, {}).get(id), screen_name: item.get(user, {}).get(screen_name), profile_image_url: item.get(user, {}).get(profile_image_url) }, reposts_count: item.get(reposts_count, 0), comments_count: item.get(comments_count, 0), attitudes_count: item.get(attitudes_count, 0), pics: WeiboParser._parse_pics(item.get(pic_infos)), # 解析图片 video_url: WeiboParser._parse_video(item.get(page_info)) # 解析视频 } # 处理长微博全文 if item.get(isLongText): # 这里理论上可以调用另一个接口获取全文需要weibo_id pass weibo_list.append(weibo) except Exception as e: logger.error(f解析单条微博时出错: {item.get(id)}, 错误: {e}) continue # 跳过这条继续解析下一条 logger.info(f成功解析 {len(weibo_list)} 条微博。) return weibo_list staticmethod def _parse_time(time_str: str) - int: 将微博时间字符串转换为时间戳。 if not time_str: return 0 # 示例格式Mon May 16 18:00:00 0800 2023 try: dt datetime.strptime(time_str, %a %b %d %H:%M:%S %z %Y) return int(dt.timestamp()) except ValueError: logger.warning(f时间格式解析失败: {time_str}) return 0 staticmethod def _parse_pics(pic_infos: Dict) - List[str]: 解析图片信息返回大图URL列表。 if not pic_infos: return [] pic_urls [] for pic_id, pic_info in pic_infos.items(): # 选择大图或原图 largest pic_info.get(largest, {}) url largest.get(url) if url: pic_urls.append(url) return pic_urls staticmethod def _parse_video(page_info: Dict) - str: 解析视频信息返回视频流URL。 if not page_info or page_info.get(type) ! video: return # 视频URL可能在media_info或urls中结构较复杂需具体分析 media_info page_info.get(media_info, {}) return media_info.get(stream_url_hd) or media_info.get(stream_url) or 关键点解析字段映射仔细分析JSON结构找到目标字段的正确路径。例如微博正文可能有多版本text,text_rawtext_raw通常是不带HTML标签的纯文本。异常处理列表解析中单条数据出错不应导致整个任务失败。用try...except包裹单条解析逻辑并记录错误。数据清洗对原始数据做初步处理如时间格式转换、来源信息清理去掉“来自”前缀、提取高质量图片/视频URL。扩展性像长微博、投票等特殊微博类型的解析可以设计成独立的方法在parse_weibo_list中根据条件调用。3.3 数据存储模块 (storage.py)解析好的数据需要持久化。根据数据量和后续用途可以选择不同的存储方式。# storage.py import csv import json import sqlite3 from pathlib import Path from typing import List, Dict, Any import logging logger logging.getLogger(__name__) class WeiboStorage: 数据存储抽象类定义接口。 def save(self, data: List[Dict], keyword: str None): raise NotImplementedError class CSVStorage(WeiboStorage): 存储到CSV文件。简单轻量适合中小规模数据。 def __init__(self, file_path: str weibo_data.csv): self.file_path Path(file_path) self._ensure_file_header() def _ensure_file_header(self): 确保CSV文件存在并写入表头。 if not self.file_path.exists(): with open(self.file_path, w, newline, encodingutf-8-sig) as f: # utf-8-sig解决Excel中文乱码 writer csv.writer(f) # 定义CSV列与parser输出的字典键对应 header [id, bid, text, created_at, timestamp, source, user_id, user_name, reposts, comments, likes, pic_urls, video_url, keyword] writer.writerow(header) def save(self, data: List[Dict], keyword: str None): if not data: return rows [] for weibo in data: # 将字典扁平化便于CSV存储 row [ weibo.get(id, ), weibo.get(bid, ), weibo.get(text, ).replace(\n, ).replace(\r, ), # 替换换行符 weibo.get(created_at, ), weibo.get(timestamp, ), weibo.get(source, ), weibo.get(user, {}).get(id, ), weibo.get(user, {}).get(screen_name, ), weibo.get(reposts_count, 0), weibo.get(comments_count, 0), weibo.get(attitudes_count, 0), |.join(weibo.get(pics, [])), # 图片URL用竖线分隔 weibo.get(video_url, ), keyword or ] rows.append(row) with open(self.file_path, a, newline, encodingutf-8-sig) as f: writer csv.writer(f) writer.writerows(rows) logger.info(f已将 {len(data)} 条微博数据追加到 {self.file_path}) class JSONStorage(WeiboStorage): 存储到JSON文件。保持数据结构方便程序读取。 def __init__(self, file_path: str weibo_data.json): self.file_path Path(file_path) self.data self._load_existing() def _load_existing(self) - List: if self.file_path.exists(): with open(self.file_path, r, encodingutf-8) as f: try: return json.load(f) except json.JSONDecodeError: return [] return [] def save(self, data: List[Dict], keyword: str None): if not data: return # 可以为每条数据添加关键词标签 if keyword: for item in data: item[_keyword] keyword self.data.extend(data) with open(self.file_path, w, encodingutf-8) as f: json.dump(self.data, f, ensure_asciiFalse, indent2) logger.info(f已将 {len(data)} 条微博数据保存到 {self.file_path}文件中共有 {len(self.data)} 条。) class SQLiteStorage(WeiboStorage): 存储到SQLite数据库。适合大规模、结构化数据管理和查询。 def __init__(self, db_path: str weibo.db): self.db_path db_path self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() # 创建微博表 cursor.execute( CREATE TABLE IF NOT EXISTS weibos ( id INTEGER PRIMARY KEY, bid TEXT UNIQUE, text TEXT, created_at TEXT, timestamp INTEGER, source TEXT, user_id TEXT, user_name TEXT, reposts_count INTEGER, comments_count INTEGER, attitudes_count INTEGER, pic_urls TEXT, video_url TEXT, keyword TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 可以创建索引以提高查询速度 cursor.execute(CREATE INDEX IF NOT EXISTS idx_timestamp ON weibos (timestamp)) cursor.execute(CREATE INDEX IF NOT EXISTS idx_keyword ON weibos (keyword)) conn.commit() conn.close() def save(self, data: List[Dict], keyword: str None): if not data: return conn sqlite3.connect(self.db_path) cursor conn.cursor() inserted_count 0 for weibo in data: try: cursor.execute( INSERT OR IGNORE INTO weibos (bid, text, created_at, timestamp, source, user_id, user_name, reposts_count, comments_count, attitudes_count, pic_urls, video_url, keyword) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( weibo.get(bid), weibo.get(text, ), weibo.get(created_at, ), weibo.get(timestamp, 0), weibo.get(source, ), weibo.get(user, {}).get(id, ), weibo.get(user, {}).get(screen_name, ), weibo.get(reposts_count, 0), weibo.get(comments_count, 0), weibo.get(attitudes_count, 0), |.join(weibo.get(pics, [])), weibo.get(video_url, ), keyword or )) if cursor.rowcount 0: inserted_count 1 except sqlite3.Error as e: logger.error(f插入数据库失败: {weibo.get(bid)}, 错误: {e}) conn.commit() conn.close() logger.info(f成功向数据库插入 {inserted_count} 条新微博记录。)关键点解析抽象与多态定义了WeiboStorage基类不同的存储方式CSV, JSON, SQLite作为子类实现。主程序只需要调用storage.save(data)无需关心底层细节。这符合“开闭原则”未来要加MySQL、MongoDB支持也很容易。CSV的细节使用utf-8-sig编码解决Excel打开中文乱码问题。将列表字段如图片URL用特定分隔符如|连接成一个字符串存储。SQLite的健壮性使用INSERT OR IGNORE避免重复插入基于bid唯一约束。创建索引可以大幅提升按时间或关键词查询的速度。数据去重在数据库层通过唯一约束实现去重是最佳实践。如果只用文件需要在解析或存储逻辑中自己判断。3.4 配置与工具模块 (config.py utils.py)这两个模块让爬虫更易用、更可靠。# config.py import os from dotenv import load_dotenv # 推荐使用python-dotenv管理环境变量 # 加载.env文件中的环境变量 load_dotenv() # 微博Cookie这是最关键的信息 # 安全提示切勿将真实的Cookie提交到Git等公开仓库建议通过环境变量读取。 WEIBO_COOKIES os.getenv(WEIBO_COOKIES, ) # 示例从文件读取更安全 # with open(cookie.txt, r) as f: # WEIBO_COOKIES f.read().strip() # 爬虫基础配置 REQUEST_DELAY (2, 5) # 请求延时范围秒 MAX_PAGES 10 # 每次爬取最大页数防止无限循环 KEYWORDS [Python, 爬虫, 人工智能] # 要搜索的关键词列表 # 存储配置 STORAGE_TYPE sqlite # 可选csv, json, sqlite DB_PATH data/weibo.db CSV_PATH data/weibo.csv JSON_PATH data/weibo.json # 代理配置如果需要 USE_PROXY False PROXY_POOL [ http://proxy1.example.com:8080, http://proxy2.example.com:8080, ]# utils.py import logging import sys from logging.handlers import RotatingFileHandler def setup_logger(name: str, log_file: str spider.log, levellogging.INFO): 设置日志记录器。 :param name: 记录器名称 :param log_file: 日志文件路径 :param level: 日志级别 :return: 配置好的logger对象 logger logging.getLogger(name) logger.setLevel(level) # 避免重复添加handler if logger.handlers: return logger # 格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) # 控制台输出 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件输出按大小轮转 file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, backupCount5 # 10MB一个文件保留5个 ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger # 在爬虫主模块中初始化日志 # logger setup_logger(__name__)关键点解析配置分离将Cookie、延时、存储路径等可变参数放在配置文件中与代码逻辑分离提高了安全性和灵活性。使用python-dotenv管理敏感信息是行业最佳实践。日志的重要性爬虫运行在后台完善的日志是排查问题的唯一依据。RotatingFileHandler可以防止日志文件无限膨胀。异常处理工具可以进一步在utils.py中添加重试装饰器、发送错误通知如邮件、钉钉等功能让爬虫更加自动化。4. 主调度器串联让爬虫跑起来有了各个模块现在需要一个“大脑”来指挥它们协同工作。主调度器spider.py负责定义爬取任务如按关键词搜索、爬取指定用户微博、控制爬取流程翻页、去重、并协调fetcher, parser, storage。# spider.py import time from typing import List, Optional from urllib.parse import quote import logging from fetcher import WeiboFetcher from parser import WeiboParser from storage import SQLiteStorage, CSVStorage, JSONStorage from config import WEIBO_COOKIES, REQUEST_DELAY, STORAGE_TYPE, DB_PATH, CSV_PATH, JSON_PATH, KEYWORDS, USE_PROXY, PROXY_POOL from utils import setup_logger logger setup_logger(__name__) class SinaWeiboSpider: def __init__(self): if not WEIBO_COOKIES: logger.error(未配置微博Cookie请在config.py或.env文件中设置WEIBO_COOKIES。) raise ValueError(Cookie未配置) self.fetcher WeiboFetcher( cookiesWEIBO_COOKIES, use_proxyUSE_PROXY, proxy_poolPROXY_POOL ) self.fetcher.min_delay, self.fetcher.max_delay REQUEST_DELAY # 根据配置选择存储方式 if STORAGE_TYPE sqlite: self.storage SQLiteStorage(DB_PATH) elif STORAGE_TYPE csv: self.storage CSVStorage(CSV_PATH) elif STORAGE_TYPE json: self.storage JSONStorage(JSON_PATH) else: logger.warning(f不支持的存储类型: {STORAGE_TYPE}默认使用SQLite。) self.storage SQLiteStorage(DB_PATH) self.parser WeiboParser() logger.info(SinaWeiboSpider 初始化完成。) def crawl_search(self, keyword: str, pages: int 10): 爬取微博关键词搜索结果。 :param keyword: 搜索关键词 :param pages: 爬取页数 logger.info(f开始爬取关键词 {keyword} 的微博计划爬取 {pages} 页。) all_weibos [] # 注意微博搜索接口URL和参数需要你通过实际分析获得以下为示例格式 base_url https://weibo.com/ajax/feed/searchHistory # 或者可能是另一个接口需要分析 # 这里假设一个接口实际需要替换 for page in range(1, pages 1): logger.info(f正在爬取第 {page} 页...) params { q: keyword, page: page, # 可能还有其他参数如type(类型), timescope(时间范围)等 } # 注意微博搜索接口可能需要不同的Headers或参数请根据实际情况调整fetcher data self.fetcher.fetch_json(base_url, paramsparams) if not data: logger.warning(f第 {page} 页请求失败或无数据可能已到末页或触发反爬。) break weibos self.parser.parse_weibo_list(data) if not weibos: logger.info(f第 {page} 页解析出0条微博停止爬取。) break all_weibos.extend(weibos) logger.info(f第 {page} 页爬取到 {len(weibos)} 条微博。) # 存储当前页数据也可以等全部爬完再存 self.storage.save(weibos, keywordkeyword) # 短暂额外延时模拟翻页 time.sleep(random.uniform(3, 6)) logger.info(f关键词 {keyword} 爬取结束共获取 {len(all_weibos)} 条微博。) return all_weibos def crawl_user_timeline(self, uid: str, pages: int 20): 爬取指定用户的时间线微博。 :param uid: 用户ID :param pages: 爬取页数 logger.info(f开始爬取用户 {uid} 的微博时间线计划爬取 {pages} 页。) all_weibos [] since_id None # 用于翻页的参数第一页为空 # 用户微博列表接口需要分析获取 base_url https://weibo.com/ajax/statuses/mymblog for page in range(1, pages 1): logger.info(f正在爬取用户 {uid} 的第 {page} 页...) params { uid: uid, page: page, feature: 0, } if since_id: params[since_id] since_id data self.fetcher.fetch_json(base_url, paramsparams) if not data: logger.warning(f用户 {uid} 第 {page} 页请求失败。) break weibos self.parser.parse_weibo_list(data) if not weibos: logger.info(f用户 {uid} 第 {page} 页解析出0条微博停止爬取。) break all_weibos.extend(weibos) self.storage.save(weibos, keywordfuser_{uid}) # 更新since_id用于下一页根据实际接口返回字段调整 since_id data.get(since_id) if not since_id: # 如果没有since_id可能用其他逻辑判断是否还有下一页 break time.sleep(random.uniform(2, 4)) logger.info(f用户 {uid} 时间线爬取结束共获取 {len(all_weibos)} 条微博。) return all_weibos def main(): 主函数演示爬虫的使用。 spider SinaWeiboSpider() # 示例1爬取多个关键词 for keyword in KEYWORDS: try: spider.crawl_search(keyword, pages5) # 每个关键词爬5页 except Exception as e: logger.error(f爬取关键词 {keyword} 时发生错误: {e}, exc_infoTrue) time.sleep(10) # 不同关键词之间长间隔 # 示例2爬取特定用户 # user_id 1234567890 # 替换为目标用户ID # spider.crawl_user_timeline(user_id, pages10) if __name__ __main__: main()关键点解析流程封装crawl_search和crawl_user_timeline方法封装了“请求-解析-存储”的完整循环并处理了翻页逻辑。翻页策略微博的翻页通常通过page参数或since_id/max_id参数控制。你需要根据实际接口确定。示例中展示了两种可能。错误隔离在main函数中每个关键词的爬取被try...except包裹这样即使一个关键词失败也不会影响其他任务。可扩展性主类结构清晰要增加新的爬取任务如爬评论、爬热搜榜只需添加新的方法即可。5. 进阶优化与实战避坑指南把上面的代码跑通你已经获得了一个可用的微博爬虫框架。但要用于生产环境还需要考虑更多细节和应对更复杂的情况。5.1 动态Cookie维护与模拟登录手动复制Cookie终究不是长久之计。我们可以尝试用Selenium或Playwright等浏览器自动化工具来模拟登录并自动获取新鲜的Cookie。# 示例使用Selenium获取Cookie (需要安装selenium和对应浏览器驱动) 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 import time def get_cookie_by_selenium(username, password): 警告此方法不稳定仅作演示。微博登录有复杂验证码。 options webdriver.ChromeOptions() options.add_argument(--headless) # 无头模式 driver webdriver.Chrome(optionsoptions) try: driver.get(https://weibo.com/login.php) # 等待并输入用户名密码实际选择器需要根据页面调整 wait WebDriverWait(driver, 10) user_input wait.until(EC.presence_of_element_located((By.NAME, username))) pass_input driver.find_element(By.NAME, password) user_input.send_keys(username) pass_input.send_keys(password) # 点击登录按钮同样需要调整选择器 login_btn driver.find_element(By.XPATH, //a[node-typesubmitBtn]) login_btn.click() # 等待登录成功跳转到主页 time.sleep(10) # 简单等待实际应用需要更智能的等待条件 # 获取Cookie cookies driver.get_cookies() # 将cookies列表转换为requests可用的字符串格式 cookie_str ; .join([f{c[name]}{c[value]} for c in cookies]) return cookie_str finally: driver.quit()重要提示微博的登录页面有强大的反自动化机制包括滑动验证码、点选验证码等。上述方法极大概率会失败仅作为思路展示。更可靠的做法是手动登录获取长期Cookie在浏览器登录后Cookie可能有效期长达数周或数月。定期手动更新一次配置文件对于低频爬取是可行的。使用已登录的Cookie池维护多个账号的Cookie池轮流使用降低单个账号的风险。考虑官方API如果数据需求量大且正规可以研究微博开放平台API虽然有限制但稳定合法。5.2 应对“未登录”或“请求频繁”的响应即使携带了Cookie服务器也可能返回提示未登录或请求频繁的JSON响应。我们需要在fetcher中增强对此类情况的判断和处理。# 在fetcher.fetch_json方法中增加状态判断 def fetch_json_enhanced(self, url: str, params: Optional[Dict] None) - Optional[Dict[str, Any]]: # ... 前面的请求代码不变 ... try: response self.session.get(url, paramsparams, proxiesproxies, timeout15) # 先检查HTTP状态码 if response.status_code 403: logger.error(访问被拒绝(403)可能是Cookie失效或IP被封禁。) # 可以在这里触发Cookie更新或代理切换逻辑 return None if response.status_code 429: logger.warning(请求过于频繁(429)需要延长延时或切换代理。) time.sleep(30) # 遇到429等待更长时间 return None response.raise_for_status() json_data response.json() # 检查微博接口特定的错误码 if json_data.get(ok) ! 1: msg json_data.get(msg, ) logger.error(f接口业务逻辑错误: {msg}) if 未登录 in msg or login in msg: logger.critical(Cookie已失效请立即更新) # 发送通知或抛出特定异常 raise CookieExpiredException(Cookie expired) return None return json_data.get(data) # ... 异常处理不变 ...5.3 增量爬取与断点续传对于需要定期更新的任务如监控某个用户的新微博我们不应该每次都从头爬取。可以利用微博ID或时间戳实现增量爬取。策略在数据库中记录已爬取的最新微博IDmax_id或时间戳。下次爬取时只请求ID大于该值或时间晚于该时间的微博。微博的since_id/max_id机制正是为此设计。在我们的crawl_user_timeline方法中已经初步体现了这个思路。更完善的方案是设计一个任务状态表记录每个爬取目标如用户ID、关键词最后成功爬取到的ID或时间并在每次爬取前读取这个状态。5.4 分布式与异步提升效率当需要爬取大量数据时单线程爬虫速度是瓶颈。异步IOasyncio aiohttp对于I/O密集型的网络请求使用异步可以极大提升吞吐量。但需要注意异步请求的频率控制更复杂更容易触发反爬。多线程/多进程可以同时爬取多个不相关的目标如不同的用户或关键词。但线程/进程间共享资源如Cookie、代理池需要加锁增加了复杂度。分布式爬虫Scrapy-Redis对于工业级应用可以使用Scrapy框架配合Redis实现多台机器协同爬取、统一任务调度和去重。这是最强大但也最复杂的方案。对于新手我建议先从稳定、可维护的单线程爬虫开始充分理解反爬策略和数据流。在确实遇到性能瓶颈且能妥善处理反爬后再考虑进阶方案。5.5 数据清洗与后续处理爬虫存储的原始数据往往需要进一步清洗才能用于分析。文本清洗去除微博正文中的表情符号如[笑cry]、话题标签#...#、用户、URL等。可以使用正则表达式。中文分词如果要做文本分析需要使用jieba等分词库。情感分析使用预训练模型如snowNLP,bert判断微博情感倾向。数据可视化用matplotlib,pyecharts等库绘制转发/评论/点赞数随时间变化的趋势图或生成词云。这些都属于爬虫下游的数据处理工作可以根据你的项目目标选择性地进行。6. 法律与伦理边界负责任地爬取数据技术是中立的但使用技术的方式有对错之分。在编写和运行爬虫时请务必牢记以下几点遵守robots.txt查看https://weibo.com/robots.txt了解微博允许和禁止爬取的路径。虽然这不是法律文件但它是网站所有者意愿的体现。尊重数据所有权微博上的内容归用户和平台所有。爬取的数据应仅用于个人学习、研究或符合合理使用原则的用途。切勿用于商业牟利、 spam、骚扰或任何违法活动。控制访问频率本文强调的延时策略不仅是技术需要也是道德要求。你的爬虫不应对微博服务器造成显著负担。保护用户隐私避免爬取和公开传播用户的敏感个人信息如手机号、身份证号、详细地址等。对公开的微博内容进行分析时也应考虑脱敏。查看服务条款仔细阅读微博的用户协议明确其关于数据抓取的条款。一个好的爬虫工程师不仅是技术高手也应是合规的践行者。用技术去创造价值而不是制造麻烦。从头开始设计和实现一个像SinaWeiboSpider这样的爬虫是一个系统工程涉及网络协议分析、数据解析、反爬对抗、系统设计等多个方面。本文提供的代码框架和思路已经涵盖了从零到一的核心环节。你可以以此为基础根据自己遇到的具体问题如接口变更、新的验证方式进行调试和扩展。在实际操作中最耗时的往往不是写代码而是分析不断变化的网页接口和反爬策略。保持耐心善用开发者工具多查资料多测试是爬虫工程师的日常。希望这篇长文能帮你少走些弯路更顺利地挖到你需要的数据。如果在实践中遇到具体问题欢迎在评论区交流讨论但记得不要分享任何敏感的Cookie或账号信息。本文还有配套的精品资源点击获取
Python微博爬虫实战：反爬策略与模块化架构设计
简介本资源是一款面向Python开发者与数据采集初学者的微博爬虫实战项目聚焦SinaWeibo平台用户画像、社交关系链及超级话题生态的数据抓取需求有效解决社交媒体公开数据获取难、反爬适配弱、结构化存储缺等实际问题。压缩包共41个文件总大小10.99MB涵盖12个核心Python源码如weibo_cn_async.py、redis_cookies.py、code_recognize.py、16个pyc字节码、2个JPG/PNG图片含SpiderFramework.jpg架构图、1个YAML账号配置、1个DLL验证码识别库yundamaAPI-x64.dll、1个可执行文件geckodriver.exe及日志、配置、属性等辅助文件模块分工明确支持登录鉴权、Cookie持久化、验证码识别与Redis缓存集成。目前已有354人学习下载提供完整可运行框架、异步多进程混合抓取逻辑、容器ID动态提取机制及logging.conf日志规范开箱即用便于二次开发与教学演示。1. 项目缘起为什么我们需要一个“聪明”的微博爬虫做舆情分析、用户画像研究或者只是想追踪某个话题的实时动态微博都是一个绕不开的数据金矿。但当你真正动手去挖的时候很快就会发现直接用requests库去请求微博页面要么拿不到数据要么拿到的是一堆乱码或者干脆被反爬机制直接“抬走”。这就是为什么我们需要一个专门为微博设计的爬虫——SinaWeiboSpider。它不是一个简单的网页下载器而是一个需要理解微博数据接口、模拟用户行为、处理动态加载和应对反爬策略的复杂工具。我最初写这个爬虫是因为一个市场分析项目需要持续追踪几个竞品官微的动态和用户反馈。手动复制粘贴不现实通用的爬虫框架又对微博这种“重防御”的站点水土不服。于是我决定基于Python从零开始构建一个稳定、高效且易于维护的微博数据采集器。这个过程踩了无数的坑也总结了不少经验。今天我就把这个项目的设计思路和核心源码拆解开来分享给大家。无论你是想学习爬虫进阶技巧还是急需一个能跑起来的微博数据采集方案这篇文章都能给你提供一条清晰的路径。2. 核心挑战解析微博的反爬机制与应对策略在动手写代码之前我们必须先搞清楚对手是谁。微博的反爬机制经过多年迭代已经相当成熟主要围绕以下几个方面构建防御。我们的爬虫设计本质上就是与这些机制进行“友好”的周旋。2.1 动态数据加载与接口分析这是第一个也是最关键的门槛。微博的网页内容尤其是时间线、评论、转发列表几乎全部通过Ajax接口动态加载。你在浏览器F12的Network面板里看到的不再是简单的HTML而是一个个XHRFetch请求返回的是结构化的JSON数据。如何应对我们的爬虫必须放弃解析HTML转向直接模拟这些Ajax请求。这意味着你需要使用浏览器开发者工具在微博页面如用户主页打开F12切换到Network网络面板筛选XHR或Fetch请求。寻找数据接口滚动页面观察新出现的请求。通常获取微博列表的接口URL会包含“containerid”和“uid”等关键参数。分析请求参数重点查看请求的Headers特别是Cookie和Authorization如果有。Query String Parameters或Payload里的参数也需要仔细研究比如since_id,page等用于翻页的参数。解析响应数据接口返回的是JSON结构清晰。你需要找到包含微博正文、发布时间、转发/评论/点赞数、用户信息等字段的路径。注意微博的接口参数和格式可能会不定期变动。一个稳定的爬虫不能硬编码接口URL最好能设计一种机制在爬虫启动时或运行中自动探测或更新接口信息虽然实现复杂但这是高阶玩法。2.2 用户身份验证Cookie与Token绝大多数有价值的微博数据接口都需要用户登录状态。爬虫如何获得这个状态Cookie这是最传统也是目前最主要的方式。用户登录后浏览器会保存一串Cookie其中包含SUB、SUBP等关键字段用于标识会话。爬虫可以直接使用这些Cookie来模拟登录状态。获取方式手动登录后从浏览器开发者工具的Application或Storage标签页中复制Cookie字符串。或者通过Selenium等自动化工具控制浏览器完成登录流程再提取Cookie。风险Cookie会过期。需要定期更新。同一个Cookie高频访问会被异常检测。Token/Authorization在一些新的或官方的API中可能会采用Bearer Token等认证方式。这通常需要更复杂的OAuth流程对于普通爬虫来说成本较高。策略选择对于个人或小规模爬取手动获取并定期更新Cookie是最快最直接的方法。我们的示例爬虫也将基于此方案。你需要将Cookie字符串保存在配置文件或环境变量中并在请求头中携带。2.3 请求频率限制与IP封禁微博会对异常高频的请求进行限制轻则返回错误码重则封禁IP一段时间。频率控制在代码中必须在每次请求之间加入随机延时。例如使用time.sleep(random.uniform(1, 3))模拟人类浏览的间隔。不要用固定间隔那太像机器了。IP代理池对于大规模、长时间爬取使用代理IP池是必须的。免费的代理不稳定付费的代理服务是更靠谱的选择。爬虫需要能够从代理池中自动获取并切换IP。User-Agent轮换使用单一的User-Agent也容易被识别。可以准备一个列表每次请求随机选择一个。2.4 数据加密与参数签名部分关键接口如发布微博、私信等的请求参数可能被加密或需要签名。这对于只做公开数据采集的爬虫来说通常不是首要问题。但如果你发现某个接口返回的数据乱码或请求总是失败可能需要检查其是否启用了加密。逆向分析JavaScript加密逻辑是爬虫工程师的高阶技能这里不展开。3. 爬虫架构设计与模块划分一个健壮的爬虫不应该把所有代码都堆在一个文件里。良好的架构能让代码更清晰、更易维护、也更容易扩展。我将这个SinaWeiboSpider设计为以下几个核心模块SinaWeiboSpider/ ├── spider.py # 爬虫主调度器负责流程控制 ├── fetcher.py # 网络请求模块封装请求逻辑和反爬策略 ├── parser.py # 数据解析模块从JSON中提取结构化信息 ├── storage.py # 数据存储模块支持保存到文件或数据库 ├── config.py # 配置文件存放Cookie、代理、关键词等 └── utils.py # 工具函数如日志、延时、异常处理3.1 网络请求模块 (fetcher.py)这是爬虫的“手”和“脚”负责与微博服务器通信。它的核心是健壮性和伪装能力。# fetcher.py import requests import time import random from typing import Optional, Dict, Any import logging logger logging.getLogger(__name__) class WeiboFetcher: def __init__(self, cookies: str, use_proxy: bool False, proxy_pool: list None): 初始化Fetcher。 :param cookies: 从浏览器复制的Cookie字符串。 :param use_proxy: 是否启用代理。 :param proxy_pool: 代理IP池列表格式如 [http://ip1:port, http://ip2:port] self.session requests.Session() # 设置请求头模拟浏览器 self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Cookie: cookies, # 关键注入Cookie Referer: https://weibo.com/ } self.session.headers.update(self.headers) self.use_proxy use_proxy self.proxy_pool proxy_pool or [] self.current_proxy_index 0 # 请求间隔配置 self.min_delay 2 self.max_delay 5 def _get_proxy(self) - Optional[Dict]: 从代理池中获取一个代理配置。 if not self.use_proxy or not self.proxy_pool: return None proxy self.proxy_pool[self.current_proxy_index] self.current_proxy_index (self.current_proxy_index 1) % len(self.proxy_pool) logger.debug(f使用代理: {proxy}) return {http: proxy, https: proxy} def _random_delay(self): 执行随机延时模拟人类操作。 delay random.uniform(self.min_delay, self.max_delay) logger.debug(f等待 {delay:.2f} 秒...) time.sleep(delay) def fetch_json(self, url: str, params: Optional[Dict] None) - Optional[Dict[str, Any]]: 发送GET请求获取JSON数据。 :param url: 接口URL :param params: 查询参数 :return: 解析后的JSON字典失败返回None self._random_delay() # 每次请求前先等待 proxies self._get_proxy() try: response self.session.get(url, paramsparams, proxiesproxies, timeout15) response.raise_for_status() # 检查HTTP状态码是否为200 # 微博接口通常返回JSON但需要检查状态码 json_data response.json() if json_data.get(ok) 1: # 微博接口常见的成功标志 return json_data.get(data) else: logger.error(f接口返回错误: {json_data.get(msg)}) return None except requests.exceptions.RequestException as e: logger.error(f网络请求失败: {url}, 错误: {e}) # 这里可以添加重试逻辑 return None except ValueError as e: logger.error(fJSON解析失败: {url}, 响应内容: {response.text[:200]}) return None # 示例获取用户主页微博列表的接口调用 # 这个接口URL和参数需要你通过分析实际网络请求获得 # url https://weibo.com/ajax/statuses/mymblog # params { # uid: 1234567890, # 用户ID # page: 1, # feature: 0 # } # fetcher WeiboFetcher(cookiesyour_cookies) # data fetcher.fetch_json(url, params)关键点解析使用Sessionrequests.Session()可以自动保持Cookie避免每次请求都手动设置。完整的Headers除了Cookie和User-AgentReferer来源页有时也很重要模拟从微博页面跳转过来的请求。异常处理网络请求充满不确定性必须用try...except包裹并记录详细的错误日志便于排查。延时与代理_random_delay和_get_proxy是反爬的核心。即使不用代理延时也必不可少。3.2 数据解析模块 (parser.py)这个模块是爬虫的“大脑”负责把fetcher拿回来的原始JSON数据转换成我们需要的、结构清晰的Python对象如字典或自定义类。# parser.py import json from datetime import datetime from typing import List, Dict, Any import logging logger logging.getLogger(__name__) class WeiboParser: 微博数据解析器。 staticmethod def parse_weibo_list(data: Dict) - List[Dict]: 解析微博列表JSON数据。 :param data: fetch_json返回的‘data’字段对应的字典。 :return: 解析后的微博字典列表。 weibo_list [] if not data or list not in data: logger.warning(数据中未找到微博列表。) return weibo_list for item in data[list]: try: weibo { id: item.get(id), # 微博ID bid: item.get(bid), # 微博bid另一种ID形式 text: item.get(text_raw), # 原始正文推荐text可能包含HTML标签 created_at: item.get(created_at), # 创建时间字符串 timestamp: WeiboParser._parse_time(item.get(created_at)), # 转为时间戳 source: item.get(source, ).replace(来自, ), # 发布来源如“iPhone客户端” user: { id: item.get(user, {}).get(id), screen_name: item.get(user, {}).get(screen_name), profile_image_url: item.get(user, {}).get(profile_image_url) }, reposts_count: item.get(reposts_count, 0), comments_count: item.get(comments_count, 0), attitudes_count: item.get(attitudes_count, 0), pics: WeiboParser._parse_pics(item.get(pic_infos)), # 解析图片 video_url: WeiboParser._parse_video(item.get(page_info)) # 解析视频 } # 处理长微博全文 if item.get(isLongText): # 这里理论上可以调用另一个接口获取全文需要weibo_id pass weibo_list.append(weibo) except Exception as e: logger.error(f解析单条微博时出错: {item.get(id)}, 错误: {e}) continue # 跳过这条继续解析下一条 logger.info(f成功解析 {len(weibo_list)} 条微博。) return weibo_list staticmethod def _parse_time(time_str: str) - int: 将微博时间字符串转换为时间戳。 if not time_str: return 0 # 示例格式Mon May 16 18:00:00 0800 2023 try: dt datetime.strptime(time_str, %a %b %d %H:%M:%S %z %Y) return int(dt.timestamp()) except ValueError: logger.warning(f时间格式解析失败: {time_str}) return 0 staticmethod def _parse_pics(pic_infos: Dict) - List[str]: 解析图片信息返回大图URL列表。 if not pic_infos: return [] pic_urls [] for pic_id, pic_info in pic_infos.items(): # 选择大图或原图 largest pic_info.get(largest, {}) url largest.get(url) if url: pic_urls.append(url) return pic_urls staticmethod def _parse_video(page_info: Dict) - str: 解析视频信息返回视频流URL。 if not page_info or page_info.get(type) ! video: return # 视频URL可能在media_info或urls中结构较复杂需具体分析 media_info page_info.get(media_info, {}) return media_info.get(stream_url_hd) or media_info.get(stream_url) or 关键点解析字段映射仔细分析JSON结构找到目标字段的正确路径。例如微博正文可能有多版本text,text_rawtext_raw通常是不带HTML标签的纯文本。异常处理列表解析中单条数据出错不应导致整个任务失败。用try...except包裹单条解析逻辑并记录错误。数据清洗对原始数据做初步处理如时间格式转换、来源信息清理去掉“来自”前缀、提取高质量图片/视频URL。扩展性像长微博、投票等特殊微博类型的解析可以设计成独立的方法在parse_weibo_list中根据条件调用。3.3 数据存储模块 (storage.py)解析好的数据需要持久化。根据数据量和后续用途可以选择不同的存储方式。# storage.py import csv import json import sqlite3 from pathlib import Path from typing import List, Dict, Any import logging logger logging.getLogger(__name__) class WeiboStorage: 数据存储抽象类定义接口。 def save(self, data: List[Dict], keyword: str None): raise NotImplementedError class CSVStorage(WeiboStorage): 存储到CSV文件。简单轻量适合中小规模数据。 def __init__(self, file_path: str weibo_data.csv): self.file_path Path(file_path) self._ensure_file_header() def _ensure_file_header(self): 确保CSV文件存在并写入表头。 if not self.file_path.exists(): with open(self.file_path, w, newline, encodingutf-8-sig) as f: # utf-8-sig解决Excel中文乱码 writer csv.writer(f) # 定义CSV列与parser输出的字典键对应 header [id, bid, text, created_at, timestamp, source, user_id, user_name, reposts, comments, likes, pic_urls, video_url, keyword] writer.writerow(header) def save(self, data: List[Dict], keyword: str None): if not data: return rows [] for weibo in data: # 将字典扁平化便于CSV存储 row [ weibo.get(id, ), weibo.get(bid, ), weibo.get(text, ).replace(\n, ).replace(\r, ), # 替换换行符 weibo.get(created_at, ), weibo.get(timestamp, ), weibo.get(source, ), weibo.get(user, {}).get(id, ), weibo.get(user, {}).get(screen_name, ), weibo.get(reposts_count, 0), weibo.get(comments_count, 0), weibo.get(attitudes_count, 0), |.join(weibo.get(pics, [])), # 图片URL用竖线分隔 weibo.get(video_url, ), keyword or ] rows.append(row) with open(self.file_path, a, newline, encodingutf-8-sig) as f: writer csv.writer(f) writer.writerows(rows) logger.info(f已将 {len(data)} 条微博数据追加到 {self.file_path}) class JSONStorage(WeiboStorage): 存储到JSON文件。保持数据结构方便程序读取。 def __init__(self, file_path: str weibo_data.json): self.file_path Path(file_path) self.data self._load_existing() def _load_existing(self) - List: if self.file_path.exists(): with open(self.file_path, r, encodingutf-8) as f: try: return json.load(f) except json.JSONDecodeError: return [] return [] def save(self, data: List[Dict], keyword: str None): if not data: return # 可以为每条数据添加关键词标签 if keyword: for item in data: item[_keyword] keyword self.data.extend(data) with open(self.file_path, w, encodingutf-8) as f: json.dump(self.data, f, ensure_asciiFalse, indent2) logger.info(f已将 {len(data)} 条微博数据保存到 {self.file_path}文件中共有 {len(self.data)} 条。) class SQLiteStorage(WeiboStorage): 存储到SQLite数据库。适合大规模、结构化数据管理和查询。 def __init__(self, db_path: str weibo.db): self.db_path db_path self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() # 创建微博表 cursor.execute( CREATE TABLE IF NOT EXISTS weibos ( id INTEGER PRIMARY KEY, bid TEXT UNIQUE, text TEXT, created_at TEXT, timestamp INTEGER, source TEXT, user_id TEXT, user_name TEXT, reposts_count INTEGER, comments_count INTEGER, attitudes_count INTEGER, pic_urls TEXT, video_url TEXT, keyword TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 可以创建索引以提高查询速度 cursor.execute(CREATE INDEX IF NOT EXISTS idx_timestamp ON weibos (timestamp)) cursor.execute(CREATE INDEX IF NOT EXISTS idx_keyword ON weibos (keyword)) conn.commit() conn.close() def save(self, data: List[Dict], keyword: str None): if not data: return conn sqlite3.connect(self.db_path) cursor conn.cursor() inserted_count 0 for weibo in data: try: cursor.execute( INSERT OR IGNORE INTO weibos (bid, text, created_at, timestamp, source, user_id, user_name, reposts_count, comments_count, attitudes_count, pic_urls, video_url, keyword) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( weibo.get(bid), weibo.get(text, ), weibo.get(created_at, ), weibo.get(timestamp, 0), weibo.get(source, ), weibo.get(user, {}).get(id, ), weibo.get(user, {}).get(screen_name, ), weibo.get(reposts_count, 0), weibo.get(comments_count, 0), weibo.get(attitudes_count, 0), |.join(weibo.get(pics, [])), weibo.get(video_url, ), keyword or )) if cursor.rowcount 0: inserted_count 1 except sqlite3.Error as e: logger.error(f插入数据库失败: {weibo.get(bid)}, 错误: {e}) conn.commit() conn.close() logger.info(f成功向数据库插入 {inserted_count} 条新微博记录。)关键点解析抽象与多态定义了WeiboStorage基类不同的存储方式CSV, JSON, SQLite作为子类实现。主程序只需要调用storage.save(data)无需关心底层细节。这符合“开闭原则”未来要加MySQL、MongoDB支持也很容易。CSV的细节使用utf-8-sig编码解决Excel打开中文乱码问题。将列表字段如图片URL用特定分隔符如|连接成一个字符串存储。SQLite的健壮性使用INSERT OR IGNORE避免重复插入基于bid唯一约束。创建索引可以大幅提升按时间或关键词查询的速度。数据去重在数据库层通过唯一约束实现去重是最佳实践。如果只用文件需要在解析或存储逻辑中自己判断。3.4 配置与工具模块 (config.py utils.py)这两个模块让爬虫更易用、更可靠。# config.py import os from dotenv import load_dotenv # 推荐使用python-dotenv管理环境变量 # 加载.env文件中的环境变量 load_dotenv() # 微博Cookie这是最关键的信息 # 安全提示切勿将真实的Cookie提交到Git等公开仓库建议通过环境变量读取。 WEIBO_COOKIES os.getenv(WEIBO_COOKIES, ) # 示例从文件读取更安全 # with open(cookie.txt, r) as f: # WEIBO_COOKIES f.read().strip() # 爬虫基础配置 REQUEST_DELAY (2, 5) # 请求延时范围秒 MAX_PAGES 10 # 每次爬取最大页数防止无限循环 KEYWORDS [Python, 爬虫, 人工智能] # 要搜索的关键词列表 # 存储配置 STORAGE_TYPE sqlite # 可选csv, json, sqlite DB_PATH data/weibo.db CSV_PATH data/weibo.csv JSON_PATH data/weibo.json # 代理配置如果需要 USE_PROXY False PROXY_POOL [ http://proxy1.example.com:8080, http://proxy2.example.com:8080, ]# utils.py import logging import sys from logging.handlers import RotatingFileHandler def setup_logger(name: str, log_file: str spider.log, levellogging.INFO): 设置日志记录器。 :param name: 记录器名称 :param log_file: 日志文件路径 :param level: 日志级别 :return: 配置好的logger对象 logger logging.getLogger(name) logger.setLevel(level) # 避免重复添加handler if logger.handlers: return logger # 格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) # 控制台输出 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件输出按大小轮转 file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, backupCount5 # 10MB一个文件保留5个 ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger # 在爬虫主模块中初始化日志 # logger setup_logger(__name__)关键点解析配置分离将Cookie、延时、存储路径等可变参数放在配置文件中与代码逻辑分离提高了安全性和灵活性。使用python-dotenv管理敏感信息是行业最佳实践。日志的重要性爬虫运行在后台完善的日志是排查问题的唯一依据。RotatingFileHandler可以防止日志文件无限膨胀。异常处理工具可以进一步在utils.py中添加重试装饰器、发送错误通知如邮件、钉钉等功能让爬虫更加自动化。4. 主调度器串联让爬虫跑起来有了各个模块现在需要一个“大脑”来指挥它们协同工作。主调度器spider.py负责定义爬取任务如按关键词搜索、爬取指定用户微博、控制爬取流程翻页、去重、并协调fetcher, parser, storage。# spider.py import time from typing import List, Optional from urllib.parse import quote import logging from fetcher import WeiboFetcher from parser import WeiboParser from storage import SQLiteStorage, CSVStorage, JSONStorage from config import WEIBO_COOKIES, REQUEST_DELAY, STORAGE_TYPE, DB_PATH, CSV_PATH, JSON_PATH, KEYWORDS, USE_PROXY, PROXY_POOL from utils import setup_logger logger setup_logger(__name__) class SinaWeiboSpider: def __init__(self): if not WEIBO_COOKIES: logger.error(未配置微博Cookie请在config.py或.env文件中设置WEIBO_COOKIES。) raise ValueError(Cookie未配置) self.fetcher WeiboFetcher( cookiesWEIBO_COOKIES, use_proxyUSE_PROXY, proxy_poolPROXY_POOL ) self.fetcher.min_delay, self.fetcher.max_delay REQUEST_DELAY # 根据配置选择存储方式 if STORAGE_TYPE sqlite: self.storage SQLiteStorage(DB_PATH) elif STORAGE_TYPE csv: self.storage CSVStorage(CSV_PATH) elif STORAGE_TYPE json: self.storage JSONStorage(JSON_PATH) else: logger.warning(f不支持的存储类型: {STORAGE_TYPE}默认使用SQLite。) self.storage SQLiteStorage(DB_PATH) self.parser WeiboParser() logger.info(SinaWeiboSpider 初始化完成。) def crawl_search(self, keyword: str, pages: int 10): 爬取微博关键词搜索结果。 :param keyword: 搜索关键词 :param pages: 爬取页数 logger.info(f开始爬取关键词 {keyword} 的微博计划爬取 {pages} 页。) all_weibos [] # 注意微博搜索接口URL和参数需要你通过实际分析获得以下为示例格式 base_url https://weibo.com/ajax/feed/searchHistory # 或者可能是另一个接口需要分析 # 这里假设一个接口实际需要替换 for page in range(1, pages 1): logger.info(f正在爬取第 {page} 页...) params { q: keyword, page: page, # 可能还有其他参数如type(类型), timescope(时间范围)等 } # 注意微博搜索接口可能需要不同的Headers或参数请根据实际情况调整fetcher data self.fetcher.fetch_json(base_url, paramsparams) if not data: logger.warning(f第 {page} 页请求失败或无数据可能已到末页或触发反爬。) break weibos self.parser.parse_weibo_list(data) if not weibos: logger.info(f第 {page} 页解析出0条微博停止爬取。) break all_weibos.extend(weibos) logger.info(f第 {page} 页爬取到 {len(weibos)} 条微博。) # 存储当前页数据也可以等全部爬完再存 self.storage.save(weibos, keywordkeyword) # 短暂额外延时模拟翻页 time.sleep(random.uniform(3, 6)) logger.info(f关键词 {keyword} 爬取结束共获取 {len(all_weibos)} 条微博。) return all_weibos def crawl_user_timeline(self, uid: str, pages: int 20): 爬取指定用户的时间线微博。 :param uid: 用户ID :param pages: 爬取页数 logger.info(f开始爬取用户 {uid} 的微博时间线计划爬取 {pages} 页。) all_weibos [] since_id None # 用于翻页的参数第一页为空 # 用户微博列表接口需要分析获取 base_url https://weibo.com/ajax/statuses/mymblog for page in range(1, pages 1): logger.info(f正在爬取用户 {uid} 的第 {page} 页...) params { uid: uid, page: page, feature: 0, } if since_id: params[since_id] since_id data self.fetcher.fetch_json(base_url, paramsparams) if not data: logger.warning(f用户 {uid} 第 {page} 页请求失败。) break weibos self.parser.parse_weibo_list(data) if not weibos: logger.info(f用户 {uid} 第 {page} 页解析出0条微博停止爬取。) break all_weibos.extend(weibos) self.storage.save(weibos, keywordfuser_{uid}) # 更新since_id用于下一页根据实际接口返回字段调整 since_id data.get(since_id) if not since_id: # 如果没有since_id可能用其他逻辑判断是否还有下一页 break time.sleep(random.uniform(2, 4)) logger.info(f用户 {uid} 时间线爬取结束共获取 {len(all_weibos)} 条微博。) return all_weibos def main(): 主函数演示爬虫的使用。 spider SinaWeiboSpider() # 示例1爬取多个关键词 for keyword in KEYWORDS: try: spider.crawl_search(keyword, pages5) # 每个关键词爬5页 except Exception as e: logger.error(f爬取关键词 {keyword} 时发生错误: {e}, exc_infoTrue) time.sleep(10) # 不同关键词之间长间隔 # 示例2爬取特定用户 # user_id 1234567890 # 替换为目标用户ID # spider.crawl_user_timeline(user_id, pages10) if __name__ __main__: main()关键点解析流程封装crawl_search和crawl_user_timeline方法封装了“请求-解析-存储”的完整循环并处理了翻页逻辑。翻页策略微博的翻页通常通过page参数或since_id/max_id参数控制。你需要根据实际接口确定。示例中展示了两种可能。错误隔离在main函数中每个关键词的爬取被try...except包裹这样即使一个关键词失败也不会影响其他任务。可扩展性主类结构清晰要增加新的爬取任务如爬评论、爬热搜榜只需添加新的方法即可。5. 进阶优化与实战避坑指南把上面的代码跑通你已经获得了一个可用的微博爬虫框架。但要用于生产环境还需要考虑更多细节和应对更复杂的情况。5.1 动态Cookie维护与模拟登录手动复制Cookie终究不是长久之计。我们可以尝试用Selenium或Playwright等浏览器自动化工具来模拟登录并自动获取新鲜的Cookie。# 示例使用Selenium获取Cookie (需要安装selenium和对应浏览器驱动) 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 import time def get_cookie_by_selenium(username, password): 警告此方法不稳定仅作演示。微博登录有复杂验证码。 options webdriver.ChromeOptions() options.add_argument(--headless) # 无头模式 driver webdriver.Chrome(optionsoptions) try: driver.get(https://weibo.com/login.php) # 等待并输入用户名密码实际选择器需要根据页面调整 wait WebDriverWait(driver, 10) user_input wait.until(EC.presence_of_element_located((By.NAME, username))) pass_input driver.find_element(By.NAME, password) user_input.send_keys(username) pass_input.send_keys(password) # 点击登录按钮同样需要调整选择器 login_btn driver.find_element(By.XPATH, //a[node-typesubmitBtn]) login_btn.click() # 等待登录成功跳转到主页 time.sleep(10) # 简单等待实际应用需要更智能的等待条件 # 获取Cookie cookies driver.get_cookies() # 将cookies列表转换为requests可用的字符串格式 cookie_str ; .join([f{c[name]}{c[value]} for c in cookies]) return cookie_str finally: driver.quit()重要提示微博的登录页面有强大的反自动化机制包括滑动验证码、点选验证码等。上述方法极大概率会失败仅作为思路展示。更可靠的做法是手动登录获取长期Cookie在浏览器登录后Cookie可能有效期长达数周或数月。定期手动更新一次配置文件对于低频爬取是可行的。使用已登录的Cookie池维护多个账号的Cookie池轮流使用降低单个账号的风险。考虑官方API如果数据需求量大且正规可以研究微博开放平台API虽然有限制但稳定合法。5.2 应对“未登录”或“请求频繁”的响应即使携带了Cookie服务器也可能返回提示未登录或请求频繁的JSON响应。我们需要在fetcher中增强对此类情况的判断和处理。# 在fetcher.fetch_json方法中增加状态判断 def fetch_json_enhanced(self, url: str, params: Optional[Dict] None) - Optional[Dict[str, Any]]: # ... 前面的请求代码不变 ... try: response self.session.get(url, paramsparams, proxiesproxies, timeout15) # 先检查HTTP状态码 if response.status_code 403: logger.error(访问被拒绝(403)可能是Cookie失效或IP被封禁。) # 可以在这里触发Cookie更新或代理切换逻辑 return None if response.status_code 429: logger.warning(请求过于频繁(429)需要延长延时或切换代理。) time.sleep(30) # 遇到429等待更长时间 return None response.raise_for_status() json_data response.json() # 检查微博接口特定的错误码 if json_data.get(ok) ! 1: msg json_data.get(msg, ) logger.error(f接口业务逻辑错误: {msg}) if 未登录 in msg or login in msg: logger.critical(Cookie已失效请立即更新) # 发送通知或抛出特定异常 raise CookieExpiredException(Cookie expired) return None return json_data.get(data) # ... 异常处理不变 ...5.3 增量爬取与断点续传对于需要定期更新的任务如监控某个用户的新微博我们不应该每次都从头爬取。可以利用微博ID或时间戳实现增量爬取。策略在数据库中记录已爬取的最新微博IDmax_id或时间戳。下次爬取时只请求ID大于该值或时间晚于该时间的微博。微博的since_id/max_id机制正是为此设计。在我们的crawl_user_timeline方法中已经初步体现了这个思路。更完善的方案是设计一个任务状态表记录每个爬取目标如用户ID、关键词最后成功爬取到的ID或时间并在每次爬取前读取这个状态。5.4 分布式与异步提升效率当需要爬取大量数据时单线程爬虫速度是瓶颈。异步IOasyncio aiohttp对于I/O密集型的网络请求使用异步可以极大提升吞吐量。但需要注意异步请求的频率控制更复杂更容易触发反爬。多线程/多进程可以同时爬取多个不相关的目标如不同的用户或关键词。但线程/进程间共享资源如Cookie、代理池需要加锁增加了复杂度。分布式爬虫Scrapy-Redis对于工业级应用可以使用Scrapy框架配合Redis实现多台机器协同爬取、统一任务调度和去重。这是最强大但也最复杂的方案。对于新手我建议先从稳定、可维护的单线程爬虫开始充分理解反爬策略和数据流。在确实遇到性能瓶颈且能妥善处理反爬后再考虑进阶方案。5.5 数据清洗与后续处理爬虫存储的原始数据往往需要进一步清洗才能用于分析。文本清洗去除微博正文中的表情符号如[笑cry]、话题标签#...#、用户、URL等。可以使用正则表达式。中文分词如果要做文本分析需要使用jieba等分词库。情感分析使用预训练模型如snowNLP,bert判断微博情感倾向。数据可视化用matplotlib,pyecharts等库绘制转发/评论/点赞数随时间变化的趋势图或生成词云。这些都属于爬虫下游的数据处理工作可以根据你的项目目标选择性地进行。6. 法律与伦理边界负责任地爬取数据技术是中立的但使用技术的方式有对错之分。在编写和运行爬虫时请务必牢记以下几点遵守robots.txt查看https://weibo.com/robots.txt了解微博允许和禁止爬取的路径。虽然这不是法律文件但它是网站所有者意愿的体现。尊重数据所有权微博上的内容归用户和平台所有。爬取的数据应仅用于个人学习、研究或符合合理使用原则的用途。切勿用于商业牟利、 spam、骚扰或任何违法活动。控制访问频率本文强调的延时策略不仅是技术需要也是道德要求。你的爬虫不应对微博服务器造成显著负担。保护用户隐私避免爬取和公开传播用户的敏感个人信息如手机号、身份证号、详细地址等。对公开的微博内容进行分析时也应考虑脱敏。查看服务条款仔细阅读微博的用户协议明确其关于数据抓取的条款。一个好的爬虫工程师不仅是技术高手也应是合规的践行者。用技术去创造价值而不是制造麻烦。从头开始设计和实现一个像SinaWeiboSpider这样的爬虫是一个系统工程涉及网络协议分析、数据解析、反爬对抗、系统设计等多个方面。本文提供的代码框架和思路已经涵盖了从零到一的核心环节。你可以以此为基础根据自己遇到的具体问题如接口变更、新的验证方式进行调试和扩展。在实际操作中最耗时的往往不是写代码而是分析不断变化的网页接口和反爬策略。保持耐心善用开发者工具多查资料多测试是爬虫工程师的日常。希望这篇长文能帮你少走些弯路更顺利地挖到你需要的数据。如果在实践中遇到具体问题欢迎在评论区交流讨论但记得不要分享任何敏感的Cookie或账号信息。本文还有配套的精品资源点击获取