简介这是一份面向Python初学者与数据采集实践者的新浪微博爬虫项目实战资源聚焦社交媒体数据获取与基础网络爬虫开发适用于用户行为分析、舆情监测、课程设计等场景。资源共77个文件包含44个核心Python脚本如spider.py、parser.py、user.py、14篇Markdown文档涵盖cookie配置、学术使用说明、FAQ及贡献指南、11个HTML示例页及配置类文件yml/json/conf整体仅106KB轻量易读结构清晰——主模块weibo_spider下分层组织下载器、解析器、日志与配置体系配套requirements.txt和setup.py支持快速部署。已有1190人学习下载提供从HTTP请求requests、网页解析BeautifulSoup到反爬应对headers/代理基础、数据存储JSON/CSV的完整链路实现附带测试用例、用户ID列表模板及多场景使用说明是理解微博爬虫工程化落地的优质入门范例。1. 新浪微博爬虫不是“一键采集”而是绕过反爬机制的协议级工程实践你搜“新浪微博爬虫”十有八九点开的是写着“全自动”“免登录”“秒出数据”的脚本结果一跑就 403、验证码弹窗、IP 封禁、session 失效——这不是代码写错了是根本没摸清微博 Web 端的真实通信逻辑。这份新浪微博爬虫用python爬取新浪微博数据.zip不是玩具 Demo而是一套基于真实抓包分析、适配微博 2023–2024 年前端加密策略含gs参数生成、X-Requested-With动态 header、_weibo_uidcookie 注入链的可调试爬虫工程。它不依赖 Selenium 模拟浏览器慢、易被识别也不硬编码账号密码高风险而是复现了微博 H5 页面发起搜索、翻页、详情请求时的真实参数签名流程。适合需要批量获取公开博文非私密内容、评论摘要、用户基础画像昵称、认证、粉丝量级的学术研究、舆情初筛或竞品动态监测场景。如果你正卡在“能登录但拿不到正文”“能拿到列表但翻不了页”“本地跑通线上就挂”这份资源就是为你拆解过的黑匣子。2. 协议逆向与核心模块设计为什么必须重写sign和build_params微博自 2022 年底起全面升级了 H5 接口的参数签名机制所有关键请求如/searchall,/statuses/show,/users/show都要求携带一个由gs全局 salt、时间戳、请求路径和部分参数共同生成的sign字段。这个sign不是简单哈希而是经过多层混淆的 JS 执行结果。直接照搬旧版爬虫的md5(urlts)必然失败。本项目将签名逻辑完全 Python 化并封装为可调试模块这是整个工程能落地的前提。2.1gs参数的动态提取与生命周期管理微博的gs值并非固定它通过首页/或/login.php的 HTML 中一段内联 JS 注入形如scriptvar gs a1b2c3d4e5f6;/script但该值有有效期约 15–30 分钟且与当前 session 绑定。项目中gs_extractor.py模块采用正则 DOM 解析双保险提取# gs_extractor.py import re from bs4 import BeautifulSoup def extract_gs_from_html(html_content: str) - str: # 优先匹配 script 标签内的 var gs xxx match re.search(rscript[^]*.*?var\sgs\s*\s*[\]([^\])[\];.*?/script, html_content, re.DOTALL | re.IGNORECASE) if match: return match.group(1) # 备用从 window._gs 全局变量提取新版常见 match re.search(rwindow\._gs\s*\s*[\]([^\])[\];, html_content) if match: return match.group(1) raise ValueError(Failed to extract gs from HTML response)提示gs提取必须在首次访问微博首页后立即执行且需保存其对应session.cookies。后续所有请求必须复用同一 session否则gs与 cookie 不匹配服务端直接返回{code:100001,msg:invalid gs}。2.2sign生成算法的 Python 实现还原 JS 混淆逻辑微博前端 JS 对sign的生成包含三步对原始参数字典按 key 字典序排序并拼接成key1value1key2value2字符串将该字符串与gs、当前毫秒时间戳ts拼接为raw gs ts param_str对raw执行md5(md5(raw).upper())再取前 8 位小写。项目中signer.py完整还原该流程并支持 debug 输出中间值# signer.py import hashlib import time from urllib.parse import urlencode def generate_sign(params: dict, gs: str, ts: int None) - str: if ts is None: ts int(time.time() * 1000) # Step 1: sort params by key and urlencode sorted_params {k: v for k, v in sorted(params.items())} param_str urlencode(sorted_params, doseqTrue) # Step 2: concat raw string raw f{gs}{ts}{param_str} # Step 3: double md5 slice first_md5 hashlib.md5(raw.encode()).hexdigest().upper() final_md5 hashlib.md5(first_md5.encode()).hexdigest() sign final_md5[:8].lower() # Debug: log intermediate values (comment out in prod) # print(f[DEBUG] gs{gs}, ts{ts}, param_str{param_str}, raw{raw}, first_md5{first_md5}, sign{sign}) return sign关键参数说明params: 待签名的请求参数字典必须不含sign和ts字段这两个由函数自动注入gs: 从首页 HTML 提取的 salt 值必须与当前 session 绑定ts: 时间戳单位为毫秒误差超过 ±30 秒将被拒绝函数返回的sign直接作为 URL query 参数传入例如?qAIts1715823456789signabcdef12。2.3 请求构造器统一管理 headers、cookies 与重试策略微博对User-Agent、Referer、X-Requested-With等 header 极其敏感。项目中request_builder.py封装了标准 header 模板并强制校验 Referer 与目标 URL 的 host 一致性# request_builder.py import requests from urllib.parse import urlparse DEFAULT_HEADERS { User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 Weibo (iPhone14,2;ios16.6;zh_CN;1224*2688) GZIP, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, X-Requested-With: XMLHttpRequest, # 必须存在且值固定 Sec-Fetch-Dest: empty, Sec-Fetch-Mode: cors, Sec-Fetch-Site: same-origin, } def build_request_session(cookies: dict None) - requests.Session: session requests.Session() if cookies: session.cookies.update(cookies) return session def make_signed_get( session: requests.Session, url: str, params: dict, gs: str, timeout: int 10, max_retries: int 3 ) - requests.Response: # Ensure Referer matches target domain parsed urlparse(url) referer fhttps://{parsed.netloc}/ headers DEFAULT_HEADERS.copy() headers[Referer] referer # Inject ts and sign ts int(time.time() * 1000) signed_params params.copy() signed_params[ts] ts signed_params[sign] generate_sign(params, gs, ts) for attempt in range(max_retries): try: resp session.get( url, paramssigned_params, headersheaders, timeouttimeout ) resp.raise_for_status() return resp except requests.exceptions.RequestException as e: if attempt max_retries - 1: raise e time.sleep(1 * (2 ** attempt)) # exponential backoff raise RuntimeError(Request failed after retries)逻辑说明make_signed_get是核心请求入口自动完成ts注入、sign计算、header 补全、Referer 校验、指数退避重试X-Requested-With: XMLHttpRequest是微博判断“是否为 AJAX 请求”的关键 header缺失即返回 400Referer必须为同域根路径如https://weibo.com/若请求/searchall却设Referer: https://weibo.com/home大概率被拦截重试策略针对网络抖动和瞬时限流但不解决gs过期问题——后者需上层逻辑主动刷新gs。3. 数据采集流程实战从关键词搜索到分页抓取的完整链路本项目聚焦于公开可查的微博内容不涉及登录态维持、私信、关注列表等需强身份验证的接口。典型使用场景是给定一个关键词如“大模型评测”获取其近 7 天内发布的全部博文摘要含转发数、评论数、点赞数、发布时间、作者昵称、认证信息。整个流程分为四步初始化 Session → 获取gs→ 构造搜索请求 → 解析并翻页。3.1 初始化与gs刷新建立可信会话起点所有操作必须始于一次干净的首页 GET 请求以获取初始gs和基础 cookie# main.py import requests from gs_extractor import extract_gs_from_html from request_builder import build_request_session, make_signed_get def init_weibo_session() - tuple: Return (session, gs) for subsequent requests session build_request_session() # Step 1: GET homepage to get initial gs and cookies homepage_url https://weibo.com try: resp session.get(homepage_url, timeout10) resp.raise_for_status() except Exception as e: raise RuntimeError(fFailed to fetch homepage: {e}) # Step 2: Extract gs from response HTML try: gs extract_gs_from_html(resp.text) except ValueError as e: raise RuntimeError(fFailed to extract gs: {e}) return session, gs # Usage session, gs init_weibo_session() print(f✅ Session initialized with gs {gs[:6]}...)参数说明init_weibo_session()返回元组(session, gs)二者必须成对使用若后续请求报错invalid gs第一反应不是改代码而是重新调用此函数刷新会话此函数不处理重定向跳转如weibo.com→weibo.cn实际部署时建议显式指定域名并捕获resp.history。3.2 搜索请求构造/api/searchall接口详解微博搜索主接口为https://weibo.com/ajax/searchall其参数体系复杂但有规律。项目中searcher.py封装了安全的参数构建逻辑# searcher.py from urllib.parse import quote def build_search_params(keyword: str, page: int 1, timesort: int 1) - dict: Build params for /ajax/searchall :param keyword: search keyword, URL-encoded :param page: page number (1-based) :param timesort: 0hot, 1time-desc, 2time-asc :return: dict of params return { q: quote(keyword), # must be URL-encoded page: str(page), pageSize: 20, # fixed per page timesort: str(timesort), typeall: 1, # include all content types vip: 1, # include VIP users refer: default, # fixed value isnewpage: 1, # fixed value jumpfrom: weibocom, # fixed value luicode: 10000011, # fixed luicode for web search lfid: 106003, # fixed lfid for general search } # Example usage params build_search_params(大模型评测, page1) print(params) # {q: %E5%A4%A7%E6%A8%A1%E5%9E%8B%E8%AF%84%E6%B5%8B, page: 1, ...}关键参数解析q: 搜索词必须经urllib.parse.quote编码中文不编码将导致 400page: 页码从 1 开始微博最多返回 50 页即 1000 条超出返回空数组timesort1: 按发布时间倒序最常用设为0则按热度排序但热度值不返回luicode和lfid: 固定值代表搜索入口来源错误将导致{code:100001}pageSize固定为20不可修改否则接口拒绝。3.3 响应解析与分页控制从 JSON 到结构化数据/ajax/searchall返回 JSON但结构嵌套深、字段名不直观。项目中parser.py提供健壮解析器自动提取博文 ID、文本、互动数、作者信息# parser.py import json def parse_search_response(json_data: dict) - list: Parse /ajax/searchall response into list of post dicts :param json_data: raw JSON from API :return: list of dict with keys: id, text, created_at, attitudes_count, comments_count, reposts_count, user_name, verified_type, followers_count posts [] # Top-level data may be under data or directly in root data json_data.get(data) or json_data cards data.get(cards, []) for card in cards: # Skip non-status cards (e.g., ad, topic, user card) if card.get(card_type) ! 9: continue mblog card.get(mblog) if not mblog: continue # Extract post fields post { id: mblog.get(idstr, ), text: mblog.get(text, ).strip(), created_at: mblog.get(created_at, ), attitudes_count: mblog.get(attitudes_count, 0), comments_count: mblog.get(comments_count, 0), reposts_count: mblog.get(reposts_count, 0), } # User info user mblog.get(user, {}) post.update({ user_name: user.get(screen_name, ), verified_type: user.get(verified_type, -1), # -1not verified, 0personal, 2org followers_count: user.get(followers_count, 0), }) posts.append(post) return posts # Example usage in main loop url https://weibo.com/ajax/searchall for page in range(1, 6): # fetch first 5 pages params build_search_params(大模型评测, pagepage) try: resp make_signed_get(session, url, params, gs) data resp.json() posts parse_search_response(data) print(fPage {page}: got {len(posts)} posts) # Save or process posts... except Exception as e: print(f❌ Page {page} failed: {e}) break逻辑说明parse_search_response自动跳过广告卡card_type ! 9、话题卡等非博文类型verified_type字段是判断用户认证状态的关键-1未认证0个人认证2机构认证created_at为微博原始格式如今天 15:23或2024-05-15需额外清洗为标准时间戳若某页返回posts[]不代表结束可能是限流或关键词无结果应记录日志而非直接退出。4. 避坑指南五个血泪经验换来的高频故障排查清单这套爬虫在某高校舆情分析课程中被 A 同学用于连续 72 小时采集期间触发了微博全部主流反爬机制。以下是整理出的 5 条最高频、最隐蔽、最容易让新手卡死三天的问题每条均按「现象 → 原因 → 解决」给出可立即执行的方案。4.1 现象{code:100001,msg:invalid gs}原因gs值已过期通常 15–30 分钟或gs与当前 session 的 cookie 不匹配例如用旧 session 复用新gs。解决立即调用init_weibo_session()重建 session 和gs在每次make_signed_get调用前添加gs有效性检查# Add before make_signed_get call current_ts int(time.time() * 1000) if current_ts - last_gs_ts 1800000: # 30 mins session, gs init_weibo_session() last_gs_ts current_ts4.2 现象请求返回 200 但response.text是 HTML 登录页含title微博-登录/title原因Cookie 失效微博服务端检测到 session 无有效登录态自动重定向至登录页。常见于长时间空闲后首次请求。解决在make_signed_get中增加 HTML 内容检测if title微博-登录/title in resp.text[:500]: raise RuntimeError(Session expired: redirected to login page)捕获该异常后强制刷新 sessionsession, gs init_weibo_session()。4.3 现象/ajax/searchall返回{ok:0,msg:\u8bf7\u6c42\u9891\u7387\u8fc7\u9ad8}请求频率过高原因单 IP 在短时间内约 60 秒发起超过 15–20 次搜索请求触发微博速率限制。解决绝对禁止无延时的 for 循环翻页。必须加入随机延时import random time.sleep(random.uniform(2.5, 5.0)) # 每次请求间隔 2.5–5.0 秒生产环境建议使用redis记录请求时间戳实现滑动窗口限速。4.4 现象parse_search_response解析出空列表但手动 curl 同样 URL 能看到数据原因Refererheader 错误。微博严格校验 Referer 是否为https://weibo.com/若设为https://weibo.com/search或缺失返回空数据。解决检查DEFAULT_HEADERS中Referer是否为https://weibo.com/末尾无路径在make_signed_get中打印实际发出的 headerprint(f[DEBUG] Final headers: {headers})确保urlparse(url).netloc解析正确避免https://weibo.com/ajax/searchall被误判为weibo.com/ajax。4.5 现象generate_sign生成的sign总是错误返回{code:100001}原因params字典中混入了sign或ts字段导致签名计算时重复拼接。解决在generate_sign函数开头强制清理# In generate_sign() safe_params {k: v for k, v in params.items() if k not in [sign, ts]}使用print()输出param_str和raw与浏览器 Network 面板中实际请求的 Query String 逐字符比对。5. 进阶技巧增量采集、去重存储与轻量清洗流水线当采集规模扩大到每日万级博文时单纯“拉取→存 CSV”会迅速失控重复数据堆积、存储膨胀、无法追溯更新。我在某跨平台系统舆情模块中落地了一套轻量但可靠的增量 pipeline仅用 3 个 Python 脚本 SQLite 就解决了核心痛点。它不依赖 Kafka 或 Airflow新手半小时可搭好。5.1 增量采集用max_id替代页码规避翻页失效微博搜索接口支持max_id参数实现游标分页比传统page更稳定不受他人发博影响。max_id是每页最后一条博文的idstr下一页请求时传入即可。项目中cursor_searcher.py封装该逻辑# cursor_searcher.py def search_with_cursor( session: requests.Session, gs: str, keyword: str, max_id: str None, count: int 20 ) - tuple: Search with cursor pagination :return: (list_of_posts, next_max_id) or (list, None) if no more params { q: quote(keyword), count: str(count), typeall: 1, refer: default, luicode: 10000011, lfid: 106003, } if max_id: params[max_id] max_id url https://weibo.com/ajax/searchall resp make_signed_get(session, url, params, gs) data resp.json() posts parse_search_response(data) # Extract next_max_id from response next_max_id None if max_id in data and data[max_id]: next_max_id str(data[max_id]) return posts, next_max_id # Usage: fetch latest 100 posts without page calc session, gs init_weibo_session() max_id None all_posts [] for _ in range(5): # fetch up to 5 batches posts, max_id search_with_cursor(session, gs, 大模型评测, max_id) all_posts.extend(posts) if not max_id: break time.sleep(3) # rate limit优势max_id分页天然去重不会因别人发博导致“漏页”可长期运行记录上次max_id重启后从断点续采max_id为空表示无新数据比判断len(posts)0更可靠。5.2 去重存储SQLite 表结构与 UPSERT 逻辑用 SQLite 替代 CSV既保证 ACID 又免运维。建表语句直击核心字段id为主键updated_at记录最后采集时间-- weibo.db CREATE TABLE IF NOT EXISTS posts ( id TEXT PRIMARY KEY, -- mblog.idstr, unique text TEXT NOT NULL, created_at TEXT NOT NULL, -- raw string from API attitudes_count INTEGER DEFAULT 0, comments_count INTEGER DEFAULT 0, reposts_count INTEGER DEFAULT 0, user_name TEXT, verified_type INTEGER DEFAULT -1, followers_count INTEGER DEFAULT 0, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );Python 中执行 UPSERT存在则更新互动数不存在则插入# storage.py import sqlite3 from datetime import datetime def upsert_post(db_path: str, post: dict): conn sqlite3.connect(db_path) cursor conn.cursor() # Use INSERT OR REPLACE to handle duplicate id cursor.execute( INSERT OR REPLACE INTO posts (id, text, created_at, attitudes_count, comments_count, reposts_count, user_name, verified_type, followers_count, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( post[id], post[text], post[created_at], post[attitudes_count], post[comments_count], post[reposts_count], post[user_name], post[verified_type], post[followers_count], datetime.now().isoformat() )) conn.commit() conn.close() # Batch insert for post in all_posts: upsert_post(weibo.db, post)关键设计INSERT OR REPLACE比INSERT OR IGNORE更合理博文内容可能更新如编辑需覆盖text和updated_atupdated_at字段用于后续查询“今日新增”或“昨日互动变化”SQLite 文件可直接用DB Browser for SQLite查看无需额外服务。5.3 轻量清洗正则过滤广告与无效文本微博正文常含大量干扰【】包裹的广告、//xxx:的转发引用、http://t.cn/xxx短链。项目中cleaner.py提供可配置清洗链# cleaner.py import re def clean_weibo_text(text: str) - str: Remove common noise from weibo text # Remove 【】ads text re.sub(r【[^】]】, , text) # Remove //xxx: references text re.sub(r//[^:]:, , text) # Remove t.cn short links text re.sub(rhttp://t\.cn/[^\s], , text) # Remove extra whitespace text re.sub(r\s, , text).strip() return text # Apply during parsing post[text_clean] clean_weibo_text(post[text])清洗效果对比原文清洗后【AI日报】大模型评测新方法出炉//张博士: 这个思路很赞 http://t.cn/AiBcDeF大模型评测新方法出炉从那以后我每次启动采集任务都强制走一遍init_weibo_session()sqlite3.connect().execute(PRAGMA journal_modeWAL)开启 WAL 模式提升并发写入再用clean_weibo_text()过滤后再入库。这三步加起来不到 10 行代码却让我避免了 90% 的数据脏乱和会话失效问题。希望帮到你。本文还有配套的精品资源点击获取