简介：QQ群爬虫项目是一套面向爬虫学习者和数据挖掘开发者的完整代码包，聚焦QQ群信息的自动抓取与解析，适合想实战网络爬虫、反爬绕过和社交平台数据采集的用户。压缩包共690个文件，以318个Python源码与316个编译后的字节码文件为主，辅以样式表、证书、表格、图片等素材，整体约4.04MB，目录结构清晰，方便按模块检索。目前已有325人学习，是一个轻量但覆盖较全的爬虫工程样例。从代码中可系统梳理网络请求、页面解析、公共接口对接、异步并发、代理轮换、异常捕获与日志记录，以及基于pandas和数据库的数据落盘方式；同时还能借鉴作者对爬虫项目的模块划分和组织方法，对独立编写、调试并部署类似采集任务有直接参考价值。
1. 从「想要一批群数据」到「跑通 QQ 群爬虫」：先把三个前提说清楚
先说结论：QQ 群爬虫真正难的不是写爬虫，而是登录态管理和频率控制。很多第一次上手的人把精力全花在解析接口返回上，结果账号被限制了一周，连客户端都登录不了，项目直接停摆。这里说的 QQ 群爬虫，指的是用程序批量获取你有权访问的 QQ 群的群资料、成员列表、消息记录和文件索引，再落成结构化数据，供社群运营分析、数据归档或可视化展示使用。
适合做这件事的人有三类：一是做社群运营需要定期盘点群规模和活跃度的；二是做数据分析需要把群聊内容变成语料或统计报表的；三是给自己管理的群做本地备份的。前提只有一个——你只爬自己有权限的群，拿到数据后不做骚扰、不传播隐私、不违反平台规则。这套边界先立住，后面所有技术选择都围绕它展开。
还有一个反直觉的判断：做 QQ 群爬虫不一定要去逆向通信协议。更稳的路线是从本地客户端缓存和用户态接口两条路选一条，前者风控压力小适合全量导出，后者灵活适合作增量更新。接下来我把这两条路的选型和落地过程完整拆开讲。
2. 两条数据通路二选一：本地方案与线上接口方案
2.1 本地方案：读取客户端落盘缓存，绕开大部分在线风控
你每天用 QQ 桌面客户端聊天，它会在本地生成一批数据文件，里面包含群列表、成员资料、最近消息记录等内容。只要客户端登录过，这些数据就真实存在磁盘上。做爬虫时直接读这些缓存文件，全程不产生任何在线请求，风控压力最小，也最适合一次导出整个群的成员和消息。
我第一次做模拟项目X时选的就是这条路。当时要导出 200 个群的成员全量数据，如果用在线接口翻页，每页几秒冷却，要跑大半天，而且中途极可能触发限制。本地缓存方案把请求压力变成了磁盘读取，十几分钟就完成了全量导出。
读取本地数据的常见做法是把缓存库当作只读 SQLite 数据库打开，先列出所有表，再定位群和成员相关的数据表。注意一个关键前提：客户端必须完全退出再读取，否则数据库文件被占用，轻则读不到最新数据，重则直接报错。
import sqlite3 from pathlib import Path # 缓存库路径来自客户端配置目录，具体位置因安装方式而异 db_path = Path(config["local_db_path"]) # 可用启动参数或配置文件指定 conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) cur = conn.cursor() # 列出全部表，先看库结构再决定取哪些字段 cur.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = [row[0] for row in cur.fetchall()] print(tables)这段代码用只读模式打开缓存库，避免程序误写数据破坏客户端文件。列出表名这一步不是多余动作，客户端版本更新后表名可能变化，先看结构再写查询，能把踩坑成本降到最低。参数方面，mode=ro是必须的，它是只读模式的开关；config["local_db_path"]建议做成配置文件项，不要写死在代码里，因为不同版本客户端的缓存路径完全不同。
本地方案的优点很突出：不触发在线频控、数据量大时速度快、可以反复重跑。缺点也同样明显——它只能拿到客户端已经同步过的数据，群里最新的聊天记录不一定都在本地；而且缓存表结构随客户端版本升级会变，需要不时重新核对字段。
2.2 在线接口方案：带登录态请求用户态接口，适合增量拉取
如果你只需要某一个群最近一天的新消息、或者某几个群的最新成员数，本地缓存往往不够新鲜。这时候走线上接口更合适：程序携带登录态请求用户态接口，拿到实时数据。
这个方案的瓶颈不在解析，而在登录态怎么来。最稳的常见做法是用自己的账号在客户端完成登录，然后从本地会话中导出请求参数，复用到爬虫请求里。所谓登录态，对在线接口来说通常是一组 Cookie 和几个请求头字段的组合，服务端靠它识别「这是一个已登录的合法客户端」。
import requests import time session = requests.Session() # 以下请求参数从本地会话导出，仅供自己有权限的群使用 session.headers.update({ "User-Agent": "Mozilla/5.0", # 保持与导出会话时一致 "Referer": f"https://{host}/", # host 取自会话初始化返回 "Cookie": config["cookie_str"], # 导出的登录态 }) params = { "group_id": 10001, # 目标群号 "start_index": 0, # 起始位置，用于翻页 "count": 50, # 每页数量，建议不超过 50 } resp = session.get(f"https://{host}/cgi-bin/group_member", params=params) data = resp.json()代码里有两个关键参数：start_index是翻页起点，和页码不是一回事，每一页结束后服务端会返回下一段的起始偏移，直接传页码大概率会拿到重复或缺失的数据；count是单页数量，不建议贪多，单页拉太多反而容易被识别为异常行为，50 是一个相对稳妥的取值。
在线接口方案适合做增量，不适合做全量。原因有两个：一是每个群、每一页的请求都算一次在线访问，量一大必定触发限流；二是接口字段中有一部分是脱敏后的占位值，拿不到真实 UIN，这对需要精确用户标识的场景是硬伤。所以我的习惯是：全量走本地，增量走在线，两条腿走路。
3. 按数据层级把「群」拆成三个抓取对象
3.1 群列表与群资料：先摸清自己手里有哪些群
拿到登录态之后，第一件事不是直接进群抓成员，而是先拉取自己的群列表。群列表是整个采集任务的地图——只有知道有哪些群、每个群的群号是什么，后续按群抓取才有目标。
群列表接口的分页参数比较特殊：它不用页码，而是用「起始位置 + 每页数量」的偏移方式。每次请求返回的 JSON 里除了当前页的群数据，还会带一个是否到底的标记，循环判断这个标记就能翻完所有群。
def fetch_group_list(session, base_params, page_size=30): all_groups = [] start = 0 while True: params = dict(base_params) params["start_index"] = start params["count"] = page_size data = session.get(f"https://{host}/cgi-bin/group_list", params=params).json() groups = data.get("group_list", []) all_groups.extend(groups) # 返回的 has_more 表示是否还有下一页 if not data.get("has_more"): break start += page_size time.sleep(1) # 冷却，不要连续翻页 return all_groups这段循环的逻辑核心是has_more标记。很多新手会写成if len(groups) < page_size: break，这在群总数刚好是 page_size 整数倍时会漏掉最后一页，或者在最后一页恰好凑满一页时会多发一次空请求。用服务端返回的标记判断更可靠。
page_size设置为 30，是因为群列表这个接口单页数量上限通常不高，而且群列表本身数量有限，没必要为省几次请求冒险调大。time.sleep(1)是给自己留的冷却间隔，在线接口方案里这是最重要的参数之一，后面避坑章节还会再提到。
拉下来的群资料字段一般包括群号、群名称、群头像 URL、当前成员数、我的群角色（群主/管理员/普通成员）。其中群号是整个后续流程的外键，所有成员和消息抓取都靠它关联。
3.2 群成员列表：全量成员、昵称与入群时间
群成员是 QQ 群爬虫最常被问到的数据。做社群运营的人都想拿到一份「群成员名册」：谁在群里、什么时候进来的、最近一次说话是什么时候。这一节数据是整个项目里最有分析价值的。
在线接口翻群成员有个独特设计：游标翻页。它不是简单地传页码，而是把上一页返回的 next 标记原样传回，服务端根据这个游标定位下一页的起始位置。游标是编码后的字符串，不要试图解析它，把它当黑匣子用就对了。
def fetch_members(session, group_id, base_params): members = [] cursor = "" seen = set() # 对成员 id 去重 while True: params = dict(base_params) params["group_id"] = group_id params["gc"] = group_id # 部分接口需要两个群标识参数 params["next"] = cursor # 游标，首页传空 data = session.get(f"https://{host}/cgi-bin/group_member", params=params).json() for m in data.get("members", []): uid = m.get("uin") or m.get("id") if uid in seen: continue seen.add(uid) members.append(m) if not data.get("has_more"): break cursor = data["next"] # 拿下一跳游标 time.sleep(1.2) return members这段代码的两个细节值得说。第一是seen去重集合：游标翻页偶尔会出现重复数据，尤其网络超时重试时，同一页会被请求两次，不去重的话最终导出量会虚高。第二是gc参数：群成员接口有时候要求同时传group_id和gc，两者值相同但语义不同，漏传gc会导致持续返回空列表，这是最有迷惑性的坑之一。
成员数据解析时你会遇到一个术语差异：接口里返回的uin是平台内部标识，它和用户日常使用的 QQ 号不一定相同。有些群成员能看到 QQ 号，有些只能拿到脱敏后的字符串，这与群设置和接口权限有关。做分析时建议以uin作为主键，因为它稳定且唯一，QQ 号当辅助字段展示用。
3.3 群消息与文件索引：决定你要备份到什么粒度
群消息是三种数据里量最大、也最容易把项目拖垮的。一个 500 人群一天的消息量可能上万条，如果全部拉下来做全量备份，接口请求次数和存储空间都是问题。所以做这一层时，先想清楚你的目标粒度：是按天增量、保留最近 N 条，还是按关键字捞特定内容。
在线消息接口的翻页机制是消息序号游标。每条消息有一个msg_seq，从某条消息开始往更早的方向翻，每次返回一页，用最早的msg_seq作为下一次请求的参数。注意方向：新消息序号大，翻页要向序号小的方向走。
def fetch_messages(session, group_id, start_seq, limit=50): msgs = [] seq = start_seq while True: params = { "group_id": group_id, "start_seq": seq, # 从这条消息开始向前翻 "count": limit, } data = session.get(f"https://{host}/cgi-bin/group_msg", params=params).json() batch = data.get("messages", []) if not batch: break # 按 msg_seq 分成新进队列和边界消息 for m in batch: if m["msg_seq"] < seq: msgs.append(m) # 边界消息防重复：等于 seq 的那条只保留一次 oldest = min(m["msg_seq"] for m in batch) if oldest >= seq: # 序号没前进，说明到底了 break seq = oldest time.sleep(1.5) return msgs这段的防重复策略是：每一次翻页都取当前批次里最小的msg_seq，下一次从它继续向前翻。边界上可能会有重复，因为上一批最后一条和下一批最前面一条可能相同，所以在清洗阶段要对msg_seq再做一次全局去重。
文件索引是另一个容易被忽略的数据。群文件列表本身不算大，一次接口请求能拿回一页，但它包含文件大小、上传者、上传时间，这些信息对判断群活跃度、识别资料分享型社群很有用。不过文件索引有一个边界：部分文件接口需要在群内有一定权限才能看完整列表，普通成员拿到的是裁剪过的子集。如果你的账号在群里只是普通成员，文件索引的完整性就要打折扣，这一点要在数据说明文档里写清楚。
4. 数据落地：三种存储格式与各自的取舍
4.1 JSON Lines：维护简单，先跑通管道再说
抓回来的数据如果直接堆在内存里，程序一退就全没了。第一个要解决的是落地格式问题。我建议第一版直接用 JSON Lines 格式，每行一条记录，追加写入。
为什么不用普通 JSON 数组？因为普通 JSON 数组在追加数据时要先把整个文件读进来、改结构、再全量写回，数据量一大就非常慢。JSON Lines 是流式追加，每抓完一页就写一行，程序中途断了，已经写入的部分还在，重跑时跳过已有记录即可。
import json from pathlib import Path def append_record(path: Path, record: dict): with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") # 示例：抓到一个成员就写一行 for m in members: append_record(Path("data/members.jsonl"), { "group_id": group_id, "uin": m.get("uin"), "nick": m.get("nick"), "join_time": m.get("join_time"), })这里的ensure_ascii=False是中文场景的必须项，不加的话所有中文会变成\uXXXX转义序列，文件能打开但人没法读。append模式天然支持断点续传，配合第五章的排查方法可以快速定位脏数据行。
JSON Lines 的缺点是查询能力弱。你要统计每个群的人数，就得把整个文件从头到尾扫一遍。所以它适合做原始数据存档和第一版跑通，不适合长期做分析。数据量到了几万条以后，就必须上数据库。
4.2 SQLite：让群数据变成可查询的本地关系库
第二版我会把数据迁到 SQLite。它不需要单独安装服务，一个文件就是整个库，随项目带着走，特别适合单机分析场景。建表时要把三张表的关系设计清楚：群表、成员表、消息表，用群号关联。
import sqlite3 conn = sqlite3.connect("qq_groups.db") cur = conn.cursor() cur.executescript(""" CREATE TABLE IF NOT EXISTS groups ( group_id INTEGER PRIMARY KEY, group_name TEXT, member_count INTEGER, my_role TEXT, updated_at TEXT ); CREATE TABLE IF NOT EXISTS members ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL, uin TEXT, nick TEXT, join_time TEXT, last_speak TEXT, UNIQUE(group_id, uin) ); CREATE TABLE IF NOT EXISTS messages ( msg_seq INTEGER, group_id INTEGER, sender_uin TEXT, content TEXT, sent_at TEXT, PRIMARY KEY (group_id, msg_seq) ); """)三张表的主键都是为去重设计的。members表的UNIQUE(group_id, uin)保证同一群的同一成员不会重复插入；messages表用(group_id, msg_seq)复合主键，天然屏蔽重放抓取带来的重复消息。后面做增量更新时，这两个唯一约束能省掉大量手工去重代码。
一个建议：不要急着在消息表上加全文索引。SQLite 的全文搜索要单独建 FTS 虚拟表，初期分析用LIKE '%关键词%'就够，等真的需要全文检索时再上 FTS5 也不迟。过早优化会让第一版变得复杂。
4.3 CSV 导出：给运营同事或可视化工具用的最终产物
数据存在 SQLite 里方便你自己查，但要把分析结果交给同事或者导入可视化工具，CSV 才是通用格式。CSV 看似简单，实际有两个常见问题：Excel 打开中文乱码、字段顺序混乱。
乱码的根源是编码。CSV 文件用 UTF-8 编码时，Excel 默认按本地编码打开，中文字符就乱成一团。解决办法是写文件时在开头加 BOM 头，Excel 看到 BOM 就会正确识别为 UTF-8。
import csv import sqlite3 conn = sqlite3.connect("qq_groups.db") cur = conn.cursor() # 按固定字段顺序查询，保证导出表格列顺序稳定 cur.execute(""" SELECT group_id, uin, nick, join_time, last_speak FROM members ORDER BY group_id, join_time """) with open("members_export.csv", "w", encoding="utf-8-sig", newline="") as f: writer = csv.writer(f) writer.writerow(["群号", "用户标识", "昵称", "入群时间", "最近发言"]) writer.writerows(cur.fetchall())utf-8-sig这个编码参数就是加 BOM 的关键写法，Python 内置支持，不需要手工写\ufeff前缀。newline=""是防止 Windows 下 CSV 行尾出现多余空行的小细节，很多人漏了它导出后表格行距异常。
CSV 的适用边界要心里有数：它能承载几十万行数据没问题，但超过百万行后 Excel 打开就会卡，那时要换成分片导出或者直接用 SQLite 配数据分析工具。CSV 是交付格式，不是存储格式。
5. 爬完一轮就翻车？QQ 群爬虫避坑清单与排查方法
5.1 登录态明明有效，请求却返回非零错误码
有一次我跑同一个接口，浏览器里打开正常，程序里请求却返回错误码，登录态明明刚导出来。查了半天发现是缺了一个请求头字段——某个接口要求额外的设备信息头，浏览器自动带了，程序里完全没传。
对策是把导出会话时看到的请求头完整复制到程序里，不要只挑 Cookie。最稳的做法是先用导出的完整请求头跑通一个请求，确认拿到正常数据后再逐步精简。缩小请求头的过程要记录，哪一次开始报错，就知道是哪个字段不能省。
5.2 翻页翻到一半，接口返回空列表或大量重复
这个现象在群成员接口最常见：前面几页正常，第三页起返回空列表；或者回看数据时发现中间有大量重复成员。
原因通常是游标参数用错了。部分接口的下一页游标藏在返回体的next字段里，而不是由前端计算「页码加一」。直接递增页码的后果就是请求了一个不存在的偏移位置，服务端返回空数据。另一个原因可能是翻页间隔太短，服务端判定异常后直接截断返回。解决方法是先检查游标字段名，再检查冷却时间，两者都正常再怀疑数据本身。
5.3 成员导出后全是 -1 或脱敏串，主键全面失效
这是本地缓存方案的高频问题。某版本客户端在缓存里对部分敏感字段做了脱敏处理，导出的成员标识全是-1或"***"，拿这些数据去做关联分析完全不可用。
这属于客户端更新带来的字段语义变化，不是代码 bug。两个解决方向：一是改走在线接口，在线接口通常能拿到完整 UIN；二是把本地库里相同位置的其他字段（比如群内昵称加最近发言时间）组合成临时主键，但可靠性有限。更实际的做法是锁定客户端版本，不在新版客户端上跑缓存导出，直到确认新表结构再说。
5.4 全量抓取当天，账号被安全验证提示拦下
这是整个项目里最值得敬畏的教训。我曾在模拟项目X里为了赶进度，把并发调到了 3 个线程、冷却压到 0.2 秒，结果半天不到，账号就弹出了异常验证，连续一周无法正常登录客户端。
现象是账号被限制，原因是短时间内集中发起大量接口请求，行为特征偏离正常用户。解决和预防的办法只有一个：收敛频率。单线程跑、每次请求间隔不低于 1 秒、单日抓取量设上限、抓到一定数量后主动停止。不要试图绕过限制，等你发现被限制时，损失已经造成了。爬虫是在跟风控赛跑，跑得慢才是跑得久。
5.5 落库数据与客户端页面显示不一致
抓回来的群人数和客户端看到的不一样，消息条数对不上。这类问题通常有三个来源：多端消息同步延迟、缓存库不是最新快照、接口返回的是近似计数而非精确值。
解决方法是建立数据时效意识：在数据库里给每张表加updated_at字段，记录这次抓取的时间；对外交付时明确说明数据截至时间点。把「数据是某个时间点的快照」写进交付说明里，比让使用者拿它跟实时页面比对要诚实得多，也能少很多解释成本。
6. 增量同步与活跃度画像：让一次爬取变成持续可用的数据管道
到这一步，你已经能跑通一轮完整的全量爬取了。但静态的一轮数据价值有限，真正好用的是一个能每天自动更新的数据管道。增量同步的核心思想是记录游标位置：每个群抓完记住当前最新的消息序号和成员变动时间点，下次只拉这个时间点之后的新数据。
def daily_increment(session, checkpoint): for group_id in checkpoint.groups(): # 只拉取上次检查点之后的新消息 new_msgs = fetch_messages( session, group_id, start_seq=checkpoint.last_seq(group_id), ) # 回到消息表做 upsert，复合主键自动去重 upsert_messages(conn, new_msgs) # 更新检查点 if new_msgs: checkpoint.set_seq(group_id, max(m["msg_seq"] for m in new_msgs)) time.sleep(1.5) # 宁可慢，不要险 checkpoint.save()这段代码把上一章的教训全部内化了：单线程、固定冷却、增量游标。checkpoint模块负责记录每个群的进度，这样程序中断后重跑不会重新抓全量历史。消息表的复合主键在upsert_messages里兜底，就算游标算错出现重叠数据，也不会写进重复记录。
增量同步稳定跑起来之后，可以做第二层价值输出：群活跃度画像。思路是把成员按发言频次和最近活跃时间分桶，识别出沉默用户、潜水用户、活跃贡献者三类人群，按周输出统计摘要。这些数据对社群运营者来说，比单纯的人数和消息总量更能说明问题。
def active_profile(conn, group_id, since_days=30): cur = conn.cursor() cur.execute(""" SELECT sender_uin, COUNT(*) AS msg_cnt, MAX(sent_at) AS last_time FROM messages WHERE group_id = ? AND sent_at >= datetime('now', ?) GROUP BY sender_uin """, (group_id, f"-{since_days} days")) for uin, cnt, last in cur.fetchall(): if cnt >= 10 and last > date_today(): tag = "活跃" elif cnt >= 3: tag = "潜水" else: tag = "沉默" update_member_tag(conn, group_id, uin, tag)这个画像逻辑用两个指标给用户打标签：发言数量和最近发言时间。阈值 10 条和 3 条不是固定的，要根据群规模调整——几百人的大群和几十人的小群，标准必然不一样。把阈值参数暴露成配置项，比写死在代码里实用得多。
最后说一下我从这个项目里学到的习惯：任何批量抓取任务都要有两套开关——总请求数上限和总运行时长上限，到点自动停止，不依赖人盯着。有一次我忘了加时长限制，晚上挂机跑增量，半夜接口异常返回了错误数据，程序进入死循环刷了几万次请求，第二天账号又进了安全保护名单。自那以后，所有任务脚本第一行就是设置停止条件，跑偏了也比跑疯了强。
数据管道不是一次搭完就结束的东西，每次客户端更新、接口调整，都要重新核对字段和游标逻辑。保持增量、保持克制、保持对风控的敬畏，这套系统就能长期稳定地给你供数。希望帮到你。
本文还有配套的精品资源，点击获取
