简介面向Python爬虫初学者的完整微博数据采集项目演示如何用requests库发送HTTP请求、BeautifulSoup解析页面并提取用户与博文信息适合想系统掌握爬虫流程、反爬策略和数据存储的开发者。资源共77个文件约106KB以44个py脚本、14个md说明、11个html页面为主另有yml、json、conf等配置依赖目录按spider、parser、downloader、writer等模块拆分结构清晰。已有1187人学习下载。项目从基础语法、HTTP协议原理讲起覆盖动态User-Agent、代理IP、验证码处理等反爬思路并结合CSV、JSON或SQLite等存储方案附带的说明文档对Cookie配置、异常处理、Scrapy扩展也有提及。读者可通过阅读源码与配套文档理解请求封装、页面解析、数据落库和异常兜底的完整链路对这种小而完整的项目类型适合用于课程设计或内部数据采集任务的实际起步模板。1. 微博数据爬虫从 Cookie 登录到百万级数据落库的完整链路前阵子帮一个调研课题抓某话题下近三个月的微博数据第一反应是找个现成的爬虫改改。翻了一圈发现大部分项目要么依赖 Selenium 挂浏览器要么登录态三天两头失效。最后用的这套方案很直接用 Python 的 requests 请求微博移动端查询接口按关键词和日期范围抓公开数据HTML 回来以后按字段拆开落到 CSV 和 MySQL全程不用渲染页面也不碰 JS 逆向。项目代码就是围绕这样一条链路组织的Cookie 管理、请求调度、HTML 解析、结果存储各占一层。适合要做舆情分析、热词趋势和样本采集的开发者也适合想弄懂小型爬虫工程该有哪些模块的人照着跑通一遍基本就理解了这类爬虫通用的骨架。2. 项目架构与设计Cookie 管理、任务调度与解析存储的基础分工2.1 目录结构模块划分与职责边界项目代码量不算大但模块边界很清晰没有把所有逻辑堆在一个 main.py 里。整个工程目录大致长这样weibo_spider/ ├── main.py # 命令行入口接收参数、组装任务 ├── config.py # 全局配置请求间隔、线程数、存储路径 ├── cookie_manager.py # Cookie 的读取、失效检测与刷新提示 ├── scheduler.py # 任务调度把关键词日期范围拆成抓取单元 ├── downloader.py # 请求下载层构造 URL、发起请求、重试 ├── parser.py # 解析层把 HTML/JSON 转成结构化字段 ├── storage.py # 存储层写入 CSV、JSON 或 MySQL ├── utils.py # 时间格式化、日志输出、字符串清理 └── requirements.txt每个文件承担的角色很单一。main.py 只做入口调度拿到命令行参数之后丢给 scheduler不直接碰请求和解析downloader 只负责发请求和收响应不管返回的内容是什么格式parser 只处理文本到字段的转换不关心数据最终落到哪里。这样分层的直接收益是替换成本低——你不想用 CSV 存数据想直接写 MySQL改 storage.py 一个文件就够了其他模块根本不用动。另一个收益是问题好定位抓过来的页面解析不出来先去 parser 层查而不是在下载层里翻代码。这里有个常见误解项目跑不起来就怪接口改了实际上大部分情况是 Cookie 失效这正是 Cookie 被拆成独立模块的原因。登录状态是这类爬虫最容易翻车的点给它独立入口后续换成扫码登录或者账号密码登录都不影响主流程。config.py 单独存在也是同一个逻辑请求间隔、线程数、存储路径这些高频调整项集中放在一起每次跑数据只改配置不动代码。2.2 任务调度关键词与日期范围如何拆分成抓取单元scheduler 不直接去请求接口而是先做任务拆分。它拿到的原始输入是关键词和日期范围如果把整个范围当成一个黑盒整体去抓中间任意一次断网或者被限制之前的数据就白抓了重新来一遍代价很高。所以项目把任务拆成“某关键词在某一天的第几页”这种最小单元按天去推进。核心逻辑大概是这样的# scheduler.py 核心逻辑把关键词时间范围拆成天粒度任务 from datetime import datetime, timedelta def build_tasks(keyword: str, start_date: str, end_date: str): tasks [] start datetime.strptime(start_date, %Y-%m-%d) end datetime.strptime(end_date, %Y-%m-%d) while start end: tasks.append({ keyword: keyword, date: start.strftime(%Y-%m-%d), page: 1, finished: False, }) start timedelta(days1) return tasks这里 keyword 是搜索关键词start_date 和 end_date 是字符串形式的日期范围。每次循环生成一个字典代表“某关键词在某一天的第 1 页”这个抓取单元。finished 字段先置为 False后续在断点续抓的时候会被改成 True标记这一天已经抓完。为什么按天拆而不是在构造 URL 时直接传一个大的时间范围两个现实考虑。第一接口对时间范围的处理经常不是精确到天的按天拆能保证单次请求的返回结果相对完整第二断点续抓方便某一天抓失败重试时只需要重跑那一天不用整个任务从头再来。你也可以在这个函数里继续扩展比如加一个 hours 参数把天拆成小时粒度适合对高峰时段的数据做更细的切片代价是任务数量会成倍增加限速策略需要同步调整。2.3 下载层与 Cookie 层一个请求从被构造到返回的完整链路任务拆好之后进入 downloader 的领地。downloader 要做的事情很单一拿到一个任务单元构造出对应的请求 URL带上配置好的请求头和 Cookie发出去再把响应内容返回。这里有一个关键实现细节微博移动端的接口有的返回 JSON有的返回 HTML所以 downloader 在返回内容之前要先判断响应类型。常见做法是看 Content-Type 头如果是 JSON 就调用 resp.json()否则就保留 resp.text# downloader.py 基础请求逻辑 def fetch(task: dict, cookie: str): url build_search_url(task[keyword], task[page]) headers { User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X), Cookie: cookie, Referer: https://m.weibo.cn/, } # 控制请求频率避免触发风控 time.sleep(random.uniform(0.5, 1.5)) resp requests.get(url, headersheaders, timeout10) if resp.status_code ! 200: return None # 非 200 一律不走解析进入重试队列 if application/json in resp.headers.get(Content-Type, ): return resp.json() return resp.textbuild_search_url 放在 utils.py 里按关键词和页码拼接参数具体实现会在第 3 章展开。这里 time.sleep 不是可有可无的即使任务拆得很细单位时间内的请求频率还是会被服务端记录。随机间隔比固定间隔更接近真实用户行为项目默认在 0.5 到 1.5 秒之间浮动。如果你只抓几万条数据这个频率问题不大要抓更大的量需要把这个间隔放到 config 里让不同场景可以单独调。Cookie 的读取和失效检测由 cookie_manager 负责。项目约定把当前有效的 Cookie 字符串复制到 cookie.txt程序启动时读入内存。每次请求返回后如果发现状态码是 401或者响应内容里出现了登录跳转标志就认为 Cookie 已经过期# cookie_manager.py Cookie 失效检测 COOKIE_FILE cookie.txt def load_cookie(): with open(COOKIE_FILE, r, encodingutf-8) as f: return f.read().strip() def is_expired(status_code: int, content: str) - bool: return status_code 401 or login in content[:300].lower()手动复制 Cookie 虽然土但稳定。用账号密码登录的方案要处理验证码和短信验证反而更容易被风控盯上。项目默认保留手动 Cookie 方案同时把 is_expired 判断做成了独立函数方便后续接扫码登录或者更自动化的登录态刷新——这是为二次开发留的口子后面第 6 章会提到怎么接。2.4 并发调度线程池消费任务队列爬虫工程的并发和普通业务并发思路不一样。普通业务并发追求每个请求都尽量快爬虫并发追求的是在单位时间内稳定地完成请求避免触发风控。项目默认是单线程顺序执行的这是保守但也稳妥的起步方式当任务量大了以后可以在 main.py 里用线程池把任务分发出去# main.py 中开启并发消费任务 from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS 4 def run_tasks(tasks, cookie): with ThreadPoolExecutor(max_workersMAX_WORKERS) as pool: futures [pool.submit(process_task, task, cookie) for task in tasks] for future in as_completed(futures): try: future.result() except Exception as e: log_error(e)process_task 必须是纯函数它只接收一个任务字典和 Cookie内部完成请求、解析、存储最后返回当天是否完成的状态。因为每个任务之间没有共享状态所以可以安全地并发。全局的请求频率要小心每个线程各自 sleep整体频率等于单线程的 MAX_WORKERS 倍这也是为什么下载间隔参数在 config.py 里单独抽出来的原因。我一般先把线程数定在 4间隔 1 秒跑十几分钟日志没有异常再往上升不靠感觉拍脑袋。3. 核心模块拆解搜索 URL 构造、JSON 解析与双通道存储3.1 搜索 URL 构造关键词编码与分页参数从 downloader 里调用的 build_search_url 开始看。移动端搜索接口最关键的参数是 containerid它决定了搜索类型type1 对应综合搜索type2 是用户搜索type3 是视频搜索。项目默认用综合搜索也就是 type1# utils.py 构造搜索 URL from urllib.parse import urlencode def build_search_url(keyword: str, page: int, search_type: int 1): params { containerid: f100103type{search_type}q{keyword}, page_type: searchall, page: page, } return https://m.weibo.cn/api/container/getIndex? urlencode(params)urlencode 来自 urllib.parse它会把中文关键词转成百分号编码也把 和 这些特殊字符正确处理掉。如果你搜的关键词里带了引号或者逗号没有这层编码请求地址很可能直接断掉或者返回空结果。page 参数控制翻页搜索接口一页大概能返回二十条左右的博文卡片具体数量要看当天的数据量不能靠页码反推总量。如果你想搜某个特定用户的内容而不是全站关键词可以在这个函数里加 if 分支把 containerid 换成用户详情页的格式。同一个 downloader 不用改只是 URL 构造函数里多了分支。这里也是接口最可能变化的地方如果某天请求返回的数据结构和预期完全不同优先抓包看接口参数而不是去解析层折腾。3.2 JSON 响应解析从卡片里抠出九个标准字段移动端搜索接口的响应是一个嵌套 JSON主体在 data 字段下面。data.cards 是卡片数组每张卡片代表一个博文展示单元卡片里的 mblog 对象才是真正的微博数据。项目在 parser.py 里用一层函数做这个转换# parser.py 卡片结构转扁平字段 def parse_card(card_obj: dict) - dict: mblog card_obj.get(mblog, {}) user mblog.get(user, {}) created_at parse_time(mblog.get(created_at, )) return { weibo_id: mblog.get(id), user_id: user.get(id), nickname: user.get(screen_name), content: mblog.get(text, ), created_at: created_at, reposts_count: mblog.get(reposts_count, 0), comments_count: mblog.get(comments_count, 0), attitudes_count: mblog.get(attitudes_count, 0), source: mblog.get(source, ), }九个字段里weibo_id 是这条微博的唯一标识后续去重靠的就是它user_id 是发博用户的数字 IDnickname 是显示名content 取自 text 字段注意这里带回 HTML 标签包括链接和表情标签存储前需要清洗created_at 是发布时间reposts_count、comments_count、attitudes_count 分别对应转发数、评论数和点赞数source 是发布设备类似客户端名称或网页版标识的包装。有一个容易忽略的点text 字段里既有纯文本也有包在 HTML 标签里的链接和表情。清洗逻辑可以作为 parse_card 的后置步骤比如用正则把标签剥掉同时把转义字符还原成可读文本。我不建议把这些逻辑直接塞进 parse_card因为字段拆分应该保持纯净清洗是独立职责。这符合前面架构分层的思路后面做二次开发时也更容易定位问题。3.3 时间解析相对时间与标准时间的换算微博返回的 created_at 是相对时间不是固定格式的时间戳。常见的几种形态是“刚刚”“5分钟前”“今天 08:30”“昨天 23:50”“10-08”这类短格式偶尔也会有完整日期。直接把这串东西丢进存储后面做趋势分析时按时间排序就是一场灾难。项目里针对这段的解析逻辑是逐级匹配# utils.py 相对时间转标准时间 import re from datetime import datetime, timedelta def parse_time(raw: str) - str: if not raw: return raw raw.strip() now datetime.now() if raw 刚刚: return now.strftime(%Y-%m-%d %H:%M:%S) if 分钟前 in raw: minutes int(re.search(r\d, raw).group()) return (now - timedelta(minutesminutes)).strftime(%Y-%m-%d %H:%M:%S) if 今天 in raw: return now.strftime(%Y-%m-%d) raw.replace(今天 , ) if 昨天 in raw: yesterday now - timedelta(days1) return yesterday.strftime(%Y-%m-%d) raw.replace(昨天 , ) if len(raw) 5 and - in raw: return f{now.year}-{raw} 00:00:00 return raw这段代码的核心是优先级匹配。这里有一个没处理完的边界情况跨年的时候“12-31 23:50”这种格式会被解析成今年的 12 月 31 日如果你抓的数据跨越了元旦那部分数据的年份就会错位。我在实际改这个项目时加了一层校验如果解析出的时间比当前时间晚了一天以上就自动把年份减一。这个坑在第 5 章会单独拎出来讲跑真实数据时真的会出现。3.4 数据落库CSV 与 MySQL 双通道存储层是独立模块项目默认支持 CSV 和 MySQL 两种通道。CSV 的写入要注意编码用 utf-8-sig 而不是 utf-8否则表格软件打开直接乱码# storage.py 写入 CSV import csv def save_to_csv(rows: list[dict], file_path: str): with open(file_path, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesrows[0].keys()) writer.writeheader() writer.writerows(rows)MySQL 通道的关键是字符集。连接参数里不写 charset或者写了 utf8 而不是 utf8mb4存进去的 emoji 表情会变成问号。项目默认用 utf8mb4同时用 executemany 批量写入来减少连接开销# storage.py MySQL 批量写入 import pymysql def save_to_mysql(rows: list[dict], conn, table_nameweibo_posts): fields list(rows[0].keys()) sql fINSERT INTO {table_name} ({,.join(fields)}) VALUES ({,.join([%s] * len(fields))}) data [tuple(r[field] for field in fields) for r in rows] cursor conn.cursor() cursor.executemany(sql, data) conn.commit()这里用字符串拼接表名和字段列表因为这个场景的字段在代码里写死不涉及用户输入所以没有问题。如果你要把这个函数开放给外部参数那就必须改成白名单校验。两条通道可以同时开CSV 用来快速预览抽样MySQL 用来存全量后面的分析脚本直接连库读。3.5 断点续抓进度文件怎么记录断点是爬虫工程里特别容易被忽略、但真正跑总量时救命的机制。项目的做法是维护一个 progress.json以“关键词日期”为键记录完成状态# utils.py 进度记录 import json, os def load_progress(): if not os.path.exists(progress.json): return {} with open(progress.json, r, encodingutf-8) as f: return json.load(f) def mark_finished(progress, key): progress[key] True with open(progress.json, w, encodingutf-8) as f: json.dump(progress, f, ensure_asciiFalse)调用时机比实现本身更重要。mark_finished 必须在数据真正落盘之后调用而不是在请求成功之后就标记。如果你把标记放在落盘之前进程在写入过程中崩了下次启动会跳过这一天直接造成数据空洞。第 5 章里有一条就是专门讲这个顺序问题。4. 从零跑通环境准备、Cookie 配置与命令行入口4.1 环境准备Python 版本与依赖安装项目用到的第三方库很少核心只有一个 requests剩下的都是辅助。requirements.txt 内容大致是这样的requests2.28.0 beautifulsoup44.11.0 lxml4.9.0 pandas1.5.0 pymysql1.0.0如果你只用 CSV 输出pymysql 可以不装直接注释掉对应行。beautifulsoup4 和 lxml 不是所有解析路径都用得上但保留着不会有坏处因为获取到的内容有 HTML 清洗的场景后面二次开发很可能用到。Python 版本建议 3.8 到 3.113.12 的 typing 语法差异在个别第三方库上还有兼容问题没必要在这个项目里给自己增加变量。安装命令就一行pip install -r requirements.txt装完以后可以先跑一句导入确认环境没有问题。这一步看着多余但能帮你把“环境问题”和“代码问题”分开后面调试的时候少走弯路。如果你有多个 Python 环境记得先激活对应的虚拟环境再装依赖否则安装位置和运行位置不一致会带来一种“明明装了却找不到模块”的玄学问题。4.2 Cookie 获取从浏览器开发者工具复制项目不提供账号密码登录Cookie 需要自己从浏览器拿。操作路径是这样用浏览器打开微博移动站登录账号然后打开开发者工具切到 Network 面板随便点开一个搜索请求在 Headers 里找到 Cookie 这一行全选复制粘贴到项目根目录的 cookie.txt 里。注意不要带上引号也不要有多余的换行一段干净的 Cookie 字符串就可以工作。这里有两点要提醒。第一复制完 Cookie 之后短时间内有效但登录态会随着账号活跃度慢慢失效。长期跑数据的话建议准备一个专门用来采集的小号平时不用它发内容只在需要的时候刷新登录尽量避免账号被风控限制。第二Cookie 里包含账号相关的标识信息不要把这个文件提交到公开仓库就算只是练手项目也应该把 cookie.txt 写进忽略列表这个习惯从项目一开始就要养成不然后悔药没地方买。4.3 命令行入口三种启动方式main.py 把参数暴露成三个维度关键词、日期范围、任务文件。最常用的启动方式如下# 方式一抓最近 7 天的数据 python main.py --keyword 新能源 --days 7 # 方式二指定精确日期范围 python main.py --keyword 新能源 --start 2025-01-01 --end 2025-01-07 # 方式三批量任务文件 python main.py --taskfile tasks.json方式一适合日常增量抓取days 参数从今天往回推不用手算日期。方式二适合补历史数据start 和 end 都是 YYYY-MM-DD 格式两个参数可以只传一个如果只传 startend 默认取今天。方式三适合一次性处理多个关键词tasks.json 里是一个数组每个元素包含 keyword、start、end 三个字段[ {keyword: 新能源, start: 2025-01-01, end: 2025-01-07}, {keyword: 人工智能, start: 2025-01-01, end: 2025-01-07} ]启动之后控制台会滚动输出日志包含当前处理的关键词、日期、页码和已抓取条数。第一次跑的时候建议只传一个关键词和一天的范围确认链路通了再扩大范围。跳过这个步骤直接全量抓一旦出问题你分不清是代码问题还是参数问题排查成本会高很多。4.4 跑通后的快速验证数据对不对要看这四个点项目跑完一批任务后先别急着去分析用下面四个检查点快速过一遍。第一个看 progress.json 里对应日期的任务是不是都标记成了 finished第二个看输出 CSV 的文件大小和行数是否和数据量匹配第三个抽查几行数据重点看 created_at 是不是标准日期格式而不是“昨天 23:50”这种相对时间第四个用 pandas 扫一下整体范围确认没有跳天和明显的异常时间import pandas as pd df pd.read_csv(output/新能源.csv) print(df.head(5)) print(时间范围:, df[created_at].min(), →, df[created_at].max()) print(总条数:, len(df))这四个检查点能过滤掉大部分采集中产生的质量问题。特别是时间范围这一项如果你要的是最近 7 天的数据但 max 时间戳比今天早了好几天说明任务调度或者时间解析某个环节出了问题这时候就该回去看日志而不是继续加量抓。5. 避坑与排查Cookie 失效、返回空页与进度记录顺序5.1 Cookie 失效但程序不退出日志一直在刷现象跑了一段时间后所有请求都返回 401但程序没有停下来日志继续一页一页地滚最后 CSV 文件里只有表头没有数据行。第一次遇到还以为接口改了费了半天劲去查 URL 构造后来才发现问题出在失效检测上。原因失效判断只检查了状态码 401 这一种情况但实际上移动端接口在登录态失效时有的请求会照常返回 200只是内容变成了登录跳转页或空壳数据。程序不知道登录已经失效任务队列照常推进数据自然落不下来。解决把失效判断分成两层。第一层看状态码第二层看响应内容连续出现三次可疑的空页面就主动终止任务而不是继续空转。同时把失效时的日志输出做得醒目些避免盯着日志看半天才发现不对。5.2 请求返回 200但 cards 数组是空的现象某一天的搜索请求成功返回 200响应结构也正常cards 数组却是空的程序认为这一天没有数据直接标记完成。原因有两类常见的触发因素。一类是关键词里带了特殊字符比如引号、逗号、百分号构造 URL 时没有被正确编码另一类是短时间内的请求频率触发了风控服务端返回了空结果来降低风险。这两种情况的表现一样都是“无数据”但原因完全相反处理方式也不一样。解决先把关键词放到浏览器里手动搜一遍确认有数据然后在构造 URL 的环节做编码检查特殊字符必须走 urlencode最后给 downloader 加上空结果计数连续出现多次空卡片时自动进入冷静期比如暂停一段时间再继续。把这三步都做了这个坑基本不会再出现。5.3 相对时间解析在跨年时错位现象抓取去年 12 月的数据时部分“12-31 23:50”被解析成了今年的 12 月 31 日 23:50时间排序全乱。你观察到的规律是越靠近跨年节点的数据时间越奇怪。原因时间解析里处理“MM-DD HH:mm”这种格式时直接使用了当前年份没有做跨年校验。12 月 31 日如果在今年还没到那它必然是去年的数据解析逻辑需要把这个推断做进去。解决在时间解析函数外面再包一层校验如果解析出的时间晚于当前时间一天以上就把年份减一。做这件事的时候注意别把正常当天数据改掉阈值条件必须严格大于一定范围而不是大于几小时就回退否则会影响真实当天的数据。5.4 进度标记过早数据直接缺失现象断点续抓之后发现某一天的 CSV 文件里只有几条数据但当时明明跑完了。检查进度文件那天的标记确实是 finished。原因进度标记放在了请求成功之后、数据落盘之前。进程如果恰好在落盘中途退出进度已经写了数据却没写全。下次启动时程序看到 finished 就跳过这一天这个空洞永远不会被补上。解决把标记完成这步挪到数据写完 CSV 或 MySQL 之后。规定顺序是拿到数据、写存储、标记完成。另外在存储层对 weibo_id 做去重即使任务重跑重复写数据也不会污染结果。从那以后我每次跑数据采集项目都强制把这块的调用顺序按这个标准走一遍。5.5 MySQL 里中文正常但表情变成问号现象CSV 文件里 emoji 正常MySQL 表格里同一段内容变成了问号。原因数据库连接字符集没有设置为 utf8mb4。旧版字符集定义不覆盖四字节的 emoji 字符只有 utf8mb4 才能完整保存。解决连接参数里显式写 charset 为 utf8mb4同时确认表结构的默认字符集也是 utf8mb4 开头。这两个地方保持一致表情和生僻字就都不会乱码。如果你用的是非关系型数据库这一步不需要但字段里的 HTML 标签清洗仍然要做。6. 二次开发从通用爬虫到可复用的定制化数据管线6.1 新增字段短链接展开与设备来源清洗项目默认已经抓了发布设备字段但它常带一层 HTML 标签包装存库之前可以加一个正则清洗把标签剥掉得到纯净的客户端名称。如果你需要进一步追踪博文里的外链可以在解析层加一个短链展开函数对内容里的短链做一次请求拿到跳转目标把目标域名作为一个新字段存下来这个字段是识别营销内容的基础素材。注意展开请求要放在下载层的限速逻辑之后否则会额外增加请求频率。6.2 营销号过滤基于行为特征而不是关键词黑名单直接过滤关键词黑名单很容易误伤也更难维护。更稳的方案是基于项目已有字段做行为判断比如发布设备集中在某个固定客户端、发文频率高于阈值、内容里短链占比偏高、评论数长期为零。这些特征都来自现有字段你只需要在存储之前加一个过滤函数把符合特征组合的样本丢弃或单独打标记就完成了基础的质量清洗。我一般会把过滤规则放进 config.py方便每次跑数据时按话题调整阈值而不是每次改代码。6.3 定时增量抓取把任务调度交给系统项目本身适合做一次性任务。如果你要长期追踪某个话题可以用系统自带的调度器每天跑一次增量任务。以类 Unix 系统自带的定时任务为例每天凌晨两点执行当天的增量抓取命令可以写成这样0 2 * * * cd /path/to/weibo_spider python main.py --keyword 新能源 --days 1增量抓取的关键仍然是去重。与其在存储层反复插入掩盖问题不如在调度层做好进度持久化progress.json 本身就是跨天重跑的锚点启动时先读进度文件跳过已经完成的任务单元而不是每次都从头抓一遍。这个项目的代码我改过两轮最大的教训是不要等项目跑挂了才想起补进度记录第一次跑之前就把断点续抓设计好后面所有增量任务都会省心很多。希望帮到你。本文还有配套的精品资源点击获取
