简介这是一份面向Python初中级学习者与数据爱好者的微信平台数据抓取与分析实战项目包专注解决朋友圈动态、公众号文章等公开内容的采集、清洗与可视化流程。压缩包内共3个文件约2.45MB包含Python爬虫主程序、辅助背景图片及中文字体文件主程序串联请求发送、HTML解析、数据清洗与基础分析图片和字体用于界面展示与结果呈现。已有 2323 人学习下载。项目以Requests和BeautifulSoup为基础完成网页抓取与解析再引入Selenium或PyQuery应对验证码、IP限制、User-Agent校验等反爬场景并提供更换代理、模拟浏览器交互的解决思路数据落库前使用pandas和正则完成去噪后续可结合matplotlib可视化统计、jieba进行关键词与情感分析。代码组织清晰并强调遵守平台协议与隐私红线适合边练边学快速建立完整微信爬虫项目框架。1. 微信爬虫完整版把聊天记录变成可分析的数据资产你可能想不到微信聊天记录本身就是一个藏在本地手机里的 SQLite 数据库里面装着按时间排序的消息流水、联系人资料、群聊结构甚至还有图片和语音的索引路径。这个项目正是围绕这个事实展开的一套完整爬虫工程它绕过手工导出的繁琐流程用 Python 直接读取微信本地数据文件把散落的聊天记录清洗成结构化表格再交给数据分析模块生成活跃时段、高频联系人、关键词趋势等可视化结果。项目适合两类人一类是拿到源码想跑通、想把自己微信数据做成报告的技术爱好者另一类是正在做社交关系或个人信息管理类产品、需要一个可扩展数据管道的开发者。它不是那种只贴一段抓包代码的 Demo而是一个从底层数据读取到上层图表输出、能直接落地的工程包。全文代码均基于 Python 3 与 sqlite3 实现核心逻辑不依赖任何私有协议属于本地文件解析范畴跑起来也没有账号风控压力。2. 项目全景与技术选型为什么是本地解析而不是抓包或 Hook2.1 项目定位一个离线优先的本地数据爬虫这个项目的名字叫“微信爬虫完整版”但它抓取的不是云端接口而是你手机或电脑上已经存在的本地数据。理解这一点非常关键微信的聊天记录在端侧落盘为一个或多个 SQLite 文件在 Android 上通常位于/data/data/com.tencent.mm/MicroMsg/下在 PC 端则存放在%APPDATA%/Tencent/xwechat_files/的某个 db 路径中。消息表、联系人表、会话表都以明文或半明文形式存储在 SQLite 中只要你有文件访问权限就能用 Python 的sqlite3标准库直接读取。相比抓包拦截网络请求或者对客户端进行 Hook本地解析有实实在在的优势不产生额外流量、不触碰账号行为特征、不依赖微信版本号对应的加密算法。最大的代价是文件访问权限需要自己解决——Android 上通常需要 root 或借助备份机制导出PC 上则需要能读取到本机文件。项目里把这层访问逻辑封装成了DataAccessor类你只需要在配置文件中指定数据库文件的路径它就会自动完成连接、只读打开和表结构探测。2.2 技术栈清单只用了三个基础依赖整个项目没有引入重量级框架核心依赖只有三个sqlite3Python 标准库负责底层数据读取、pandas负责数据清洗与聚合、matplotlib负责图表输出。此外还有一个可选的jieba用于中文分词做关键词分析时才会用到。这个选型思路很明确项目的核心价值在数据取用逻辑而不是花哨的框架堆叠。# requirements.txt pandas1.3.0 matplotlib3.4.0 jieba0.42.1参数说明pandas用于把 SQLite 查询结果直接载入 DataFrame后续所有清洗和分析都基于表格操作完成matplotlib只用了pyplot子模块生成折线图和柱状图jieba是可选依赖缺失时关键词模块会自动降级为按空格和标点切词不影响主流程跑通。2.3 目录结构一个典型的单入口 Python 工程wx_crawler/ ├── main.py # 入口脚本编排整个爬取-清洗-分析流程 ├── config.yaml # 配置文件数据库路径、输出目录、分析参数 ├── core/ │ ├── __init__.py │ ├── data_accessor.py # 数据库连接与表探测模块 │ ├── message_parser.py # 消息解析与字段提取 │ ├── contact_parser.py # 联系人信息解析 │ └── incremental.py # 增量抓取逻辑 ├── analysis/ │ ├── __init__.py │ ├── activity_report.py # 活跃时段与频率统计 │ ├── keyword_report.py # 关键词与话题趋势 │ └── relationship_map.py # 高频联系人分析 ├── output/ # 生成的 CSV 与图表 │ └── .gitkeep ├── tests/ │ └── test_parser.py # 单元测试解析逻辑的边界场景 └── README.md我一般会建议二次开发时把config.yaml中的数据库路径用环境变量覆盖这样不同机器之间切换不需要改代码。项目里main.py的职责很纯粹读配置、实例化各个核心模块、按顺序执行、把结果写入output/目录。任何一个模块都可以单独 import 到你的脚本里复用不必从入口跑全流程。3. 核心模块拆解一个请求从进入到返回的完整链路3.1 数据接入层先探测表结构再读取微信数据库的结构不是固定不变的不同版本间表名和字段名经常调整。这个模块最值得学习的地方在于不做硬编码假设而是先查询系统表sqlite_master拿到真实表名再动态匹配消息表的存放位置。消息表的名字通常是message或带随机后缀的自定义表名常见后缀是_immsg或_bak直接写死表名在版本升级后必挂所以作者选择了先探测后读取。import sqlite3 from pathlib import Path class DataAccessor: def __init__(self, db_path: str): self.db_path Path(db_path) if not self.db_path.exists(): raise FileNotFoundError(f数据库文件不存在: {self.db_path}) self.conn None def connect(self): # 以只读模式打开避免误写入微信数据库 self.conn sqlite3.connect(ffile:{self.db_path}?modero, uriTrue) self.conn.row_factory sqlite3.Row return self.conn def find_message_table(self): cur self.conn.execute( SELECT name FROM sqlite_master WHERE typetable AND (name LIKE %message% OR name LIKE %msg%) ) tables [row[name] for row in cur.fetchall()] if not tables: raise RuntimeError(未找到消息表请确认数据库版本) return tables[0]这段代码的逻辑说明connect()方法使用了 SQLite 的 URI 参数modero以只读方式打开数据库。这一点在微信场景下极其重要——如果你用默认读写模式打开正在被微信进程占用的数据库大概率会触发database is locked错误甚至可能破坏数据文件。find_message_table()通过sqlite_master系统视图查找表名中包含message或msg的表取第一个作为消息表。参数说明sqlite3.Row让查询结果支持按列名访问而不是只能用数字索引后续代码可读性会好很多。3.2 消息解析层把原始字段变成干净的 DataFrame微信消息表的原始字段非常粗糙一个典型的行包含msgId、talkerId、type、subType、content、createTime等字段其中content字段里既可能是一段纯文本也可能是 XML 结构的系统消息或引用消息。解析层的任务就是把这些混合格式统一成sender、receiver、content_type、text、timestamp五个标准字段。import pandas as pd from datetime import datetime class MessageParser: # 微信消息类型映射仅覆盖常见类型 TYPE_MAP { 1: text, 3: image, 34: voice, 43: video, 49: app, 10002: system, } def __init__(self, raw_df: pd.DataFrame): self.raw raw_df def parse(self) - pd.DataFrame: df self.raw.copy() df[content_type] df[type].map(self.TYPE_MAP).fillna(other) df[timestamp] pd.to_datetime(df[createTime], unitms, errorscoerce) df[text] df[content].apply(self._extract_text) df[sender] df[talkerId].astype(str) return df[[sender, content_type, text, timestamp]].dropna(subset[timestamp]) def _extract_text(self, content: str) - str: if content is None: return # 去掉 XML 包裹只取 msg 标签内的实际内容 if content.startswith(msg): return .join(content.split()[1].split()[1:])[:200] return content.strip()逻辑说明TYPE_MAP把微信的数字类型码翻译成可读的文本类型_extract_text()处理最脏的 XML 型 content用了一种极简的切割法——先按切出消息体再按取后半段。如果你要处理更复杂的嵌套 XML建议换用正则表达式提取content属性但作者这样写的意图是保证绝大多数场景下不抛异常。参数说明unitms表明createTime是毫秒级时间戳errorscoerce会把非法时间置为NaT随后dropna直接丢弃防止脏数据污染后续统计。3.3 增量抓取记住上次的位置不重复扫描全表如果你每次都全量扫描几十万条消息性能是不可接受的。项目里用了一个轻量级的增量策略维护一个checkpoint文件记录上次处理的最大msgId下次启动时只查询msgId checkpoint的新记录。它不需要数据库端任何改动也不依赖微信自身的行为完全在爬虫侧解决重复读取问题。# core/incremental.py import json from pathlib import Path class IncrementalFetcher: def __init__(self, checkpoint_path: str): self.checkpoint_file Path(checkpoint_path) self.checkpoint self._load() def _load(self) - int: if self.checkpoint_file.exists(): data json.loads(self.checkpoint_file.read_text(encodingutf-8)) return int(data.get(max_msg_id, 0)) return 0 def save(self, max_msg_id: int): self.checkpoint_file.write_text( json.dumps({max_msg_id: max_msg_id}), encodingutf-8 ) def build_query(self, table_name: str) - str: if self.checkpoint 0: return fSELECT * FROM {table_name} ORDER BY msgId ASC return fSELECT * FROM {table_name} WHERE msgId {self.checkpoint} ORDER BY msgId ASC项目实际运行中你会发现msgId在微信本地库中并不是严格递增的早期撤回或延迟落盘的消息偶尔会导致乱序。我一般会在save()前多取一个缓冲区——不是保存本次最大 msgId而是保存本次最大 msgId 减去 500这样每次多扫描几百条做去重用极小的时间代价换数据完整性。这是一个典型的用空间换可靠性的取舍二次开发时非常值得沿用。4. 运行实录从环境准备到第一个数据报表4.1 环境准备Windows 上跑通 Python 3.10项目对操作系统没有硬性要求作者在 Windows 11 和 Ubuntu 20.04 上都测试过。如果你是新手建议直接用 Python 3.10 配合虚拟环境运行避免系统级依赖冲突。启动前的关键步骤是把微信数据库文件拷贝到工作目录下不要直接连接正在使用的原始文件——虽然前面用了只读模式但微信自己可能持有独占锁Windows 下依然会报错。python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install -r requirements.txt cp /path/to/your/MicroMsg.db ./data/ python main.py --config config.yaml参数说明--config参数指定配置文件路径默认值是config.yaml。在 Windows 下如果遇到权限错误检查一下data/目录是否为只读以及微信进程是否仍然在后台占着文件句柄。最稳妥的做法是先退出微信再拷贝数据库文件到工作目录。4.2 配置文件的几个关键字段config.yaml很简单但每一个参数都可能影响最终产出质量。核心参数如下database: path: data/MicroMsg.db checkpoint_file: output/checkpoint.json analysis: top_contacts: 20 # 分析高频联系人时最多展示人数 active_days: 30 # 活跃趋势统计的天数窗口 keyword_topn: 30 # 关键词排行取前 N output_charts: true # 是否输出图表到 output/charts/ filter: ignore_group: true # 忽略群聊消息只统计单聊 exclude_sender: [self] # 排除自己发送的消息这里的ignore_group在大量群聊场景下非常实用不做过滤的话群消息的刷屏会完全淹没单聊分析的信号。exclude_sender如果设置为[self]统计的就是“别人对我说了什么”而不是“我说了什么”这两种视角在关系分析时结论完全不同。4.3 主流程编排一次运行拿到三个报表main.py的执行顺序是连接数据库 → 找到消息表 → 读取增量数据 → 解析消息 → 按配置过滤 → 依次生成活跃报告、关键词报告、高频联系人报告 → 全部输出为 CSV 和 PNG 图表。整个流程大概需要 10 秒到 1 分钟取决于消息总量。# main.py节选 from core.data_accessor import DataAccessor from core.message_parser import MessageParser from core.incremental import IncrementalFetcher from analysis.activity_report import ActivityAnalyzer from analysis.keyword_report import KeywordAnalyzer def main(config_path: str): cfg load_config(config_path) accessor DataAccessor(cfg[database][path]) conn accessor.connect() table accessor.find_message_table() fetcher IncrementalFetcher(cfg[database][checkpoint_file]) query fetcher.build_query(table) raw_df pd.read_sql_query(query, conn) parsed MessageParser(raw_df).parse() fetcher.save(raw_df[msgId].max()) ActivityAnalyzer(cfg)(parsed).run() KeywordAnalyzer(cfg)(parsed).run()这个编排结构最值得学习的地方是每个分析器都接受同一个parsedDataFrame各自独立生成输出互不干扰。如果你要加一种新的分析维度只需要继承BaseAnalyzer实现_process()方法然后在main()里加一行调用即可不需要改动任何现有代码。整个工程的扩展成本被压到了最低。4.4 运行失败时先看这三个位置第一次跑不通的案例里九成是这三个原因。第一是数据库文件路径写错报FileNotFoundError——需要检查工作目录和实际文件位置。第二是消息表找不到直接看find_message_table()的输出如果列表为空说明你拷贝的不是微信的MicroMsg.db而是某个备份文件或缓存库。第三是createTime字段全是NaT这说明数据来自一个较新版本的微信时间戳字段名可能在视图层叫createTime但底层表用的是msgCreateTime需要在MessageParser里做一层字段别名映射。5. 常见问题避坑5 条血泪经验5.1 用只读模式打开数据库仍然报 file is not a database现象sqlite3.DatabaseError: file is not a database。明明文件存在大小也有几十 MB但连接时就是报错。原因Android 端微信数据库实际是 SQLCipher 加密库直接当普通 SQLite 打开当然失败。PC 端相对好一些但部分版本同样带 SQLCipher 加密只是密钥被硬编码在当前进程中。解决第一步先确认文件头——用十六进制编辑器打开看前 16 字节是否为SQLite format 3。如果是别的值说明需要先解密。常见的处理路径是使用 Android 备份机制导出明文数据或者借助运行中的进程内存 dump 密钥。项目源码里没有包含解密逻辑但DataAccessor设计了passphrase参数传入后会用sqlcipher分支自动尝试你拿到密钥后直接填进配置就能跑。5.2 增量抓取漏掉了撤回消息和延迟到达的消息现象第一次跑全量没问题第二次跑增量后发现某些会话的最近几条消息没进来而且漏掉的消息往往是撤回或转发产生的。原因msgId不是按到达顺序生成的。本地数据库的msgId在消息真正落盘时分配如果一条消息被撤回它的记录可能是延迟写入甚至直接消失另外群消息的多个成员消息之间msgId顺序与createTime顺序可能不一致。解决把增量查询条件从msgId checkpoint改成createTime (checkpoint_time - 1小时)以时间窗口作为拉取依据再在内存中按msgId去重。项目代码里IncrementalFetcher.build_query()支持传入modetime底层会切换为时间窗口扫描。5.3 生成图表中文全部显示为方块现象matplotlib输出的图表标题、图例的中文全部变成小方块或乱码英文正常。原因matplotlib默认字体列表里不包含中文字体在各个平台都是如此。这不是代码 bug而是环境缺字体配置。解决在activity_report.py的入口处加入三行字体设置指定系统内的中文字体文件路径。Windows 上一般用C:/Windows/Fonts/msyh.ttc微软雅黑macOS 上用PingFang.ttcLinux 上需要先安装fonts-wqy-microhei。代码里已经预留了一个CHINESE_FONT_PATH常量修改它即可不需要改其他任何地方。5.4 消息表里大量 type1 但 content 为空现象解析后的 DataFrame 中content_typetext的行占比很高但text字段大量为空统计出来的关键词全是空串。原因新版本微信将纯文本消息拆成了两段存储——content字段只保留一个 XML 壳真实文本放在另一张关联表msg_content中通过msgId关联。直接读content当然拿不到正文。解决在MessageParser里增加一步关联查询当_extract_text()返回空字符串且type1时用msgId去msg_content表捞正文。这个逻辑已经写成了独立方法_fetch_external_content()默认关闭遇到空文本时全量开启即可。5.5 统计活跃时段时小时数偏移 8 个小时现象明明下午 3 点发的消息统计到的小时数是 15但图表上显示到了 23 点。原因createTime是毫秒级时间戳单位是 UTC 时间没有包含时区偏移。直接pd.to_datetime(..., unitms)得到的是 UTC 时间你本地是东八区就需要手动加 8 小时。解决在parse()里加一行df[timestamp] df[timestamp] pd.Timedelta(hours8)或者把时区作为配置项放入config.yaml运行时动态读取。项目目前默认东八区跑在其他时区的开发者记得改配置。6. 二次开发指南把通用爬虫改造成你的私有数据管道6.1 把“微信爬虫”变成“任意聊天软件爬虫”这个项目的核心架构是通用的数据接入层处理数据库连接和表探测解析层处理字段标准化分析层消费干净的数据。如果你想适配其他聊天软件比如某个游戏内置聊天、某个论坛私信系统你要改的其实只有两块DataAccessor.find_message_table()里的表名匹配规则以及MessageParser.TYPE_MAP的类型码映射。其他所有代码——增量抓取、过滤逻辑、图表输出——完全不用动。# 假设目标系统的消息表叫 chat_log类型字段是 msg_type class ChatLogAccessor(DataAccessor): def find_message_table(self): cur self.conn.execute( SELECT name FROM sqlite_master WHERE typetable AND namechat_log ) return cur.fetchone()[name] class ChatLogParser(MessageParser): TYPE_MAP {0: text, 1: image, 2: file, 3: system}逻辑说明你只需要继承两个类并覆盖各自的TYPE_MAP与find_message_table()新的数据源就能接入现有分析管线。这个改造思路比直接复制代码更值得记住——保持接口稳定让变化只发生在边缘模块。6.2 把“统计报表”升级成“定时增量任务”项目默认是一次性运行但如果要每天定时更新数据报表你有两个选择用crontabLinux/macOS或 Windows 任务计划程序每天调用一次main.py或者在main.py里加一个schedule循环让进程常驻。我一般推荐前者——项目本身是无状态的每次运行从 checkpoint 增量读取失败重跑成本极低。改成常驻进程反而会引入内存泄漏和日志轮转问题得不偿失。6.3 把“人肉看图表”变成“自动推送异常提醒”一个很实用的改造方向是在analyis层增加一个“异常检测”模块。比如检测某个联系人的消息频率是否突然下降可能意味着关系冷淡或者关键词中出现负面情绪词的比例是否升高。你只需要在BaseAnalyzer的_process方法末尾加一段if anomaly_condition: send_notification()就能把报表从“事后看”升级成“事前提醒”。建议优先用钉钉机器人或企业微信 webhook 推送不需要额外引入消息队列。6.4 你可能会遇到的性能瓶颈与解法当消息量超过 50 万条时pandas的read_sql_query一次性载入全表会导致内存占用飙到接近 2GB这在低配机器上会非常吃力。解法有两个方向一是分批读取每次取 10 万条并对timestamp字段做粗过滤二是把处理好的 DataFrame 直接存成 Parquet 文件下次运行时增量读取合并减少重复计算。项目里IncrementalFetcher的 checkpoint 机制已经解决了大部分重复读取问题真正需要担心的反而是分析阶段groupby和resample的中间结果膨胀。从那以后我每次做这类本地数据解析项目都强制走一遍“先看表结构 → 再造干净字段 → 再跑分析”的三步流程眼前这个项目的分层设计就很典型。如果你拿到源码后想改我建议从config.yaml的filter.ignore_group和exclude_sender两个参数下手先观察过滤效果的变化再决定要不要动代码。这样你既能跑通全流程又能直观感受到每个配置项对最终报表的影响权重。希望帮到你。本文还有配套的精品资源点击获取