简介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(ffile:{db_path}?modero, uriTrue) cur conn.cursor() # 列出全部表先看库结构再决定取哪些字段 cur.execute(SELECT name FROM sqlite_master WHERE typetable) tables [row[0] for row in cur.fetchall()] print(tables)这段代码用只读模式打开缓存库避免程序误写数据破坏客户端文件。列出表名这一步不是多余动作客户端版本更新后表名可能变化先看结构再写查询能把踩坑成本降到最低。参数方面modero是必须的它是只读模式的开关config[local_db_path]建议做成配置文件项不要写死在代码里因为不同版本客户端的缓存路径完全不同。本地方案的优点很突出不触发在线频控、数据量大时速度快、可以反复重跑。缺点也同样明显——它只能拿到客户端已经同步过的数据群里最新的聊天记录不一定都在本地而且缓存表结构随客户端版本升级会变需要不时重新核对字段。2.2 在线接口方案带登录态请求用户态接口适合增量拉取如果你只需要某一个群最近一天的新消息、或者某几个群的最新成员数本地缓存往往不够新鲜。这时候走线上接口更合适程序携带登录态请求用户态接口拿到实时数据。这个方案的瓶颈不在解析而在登录态怎么来。最稳的常见做法是用自己的账号在客户端完成登录然后从本地会话中导出请求参数复用到爬虫请求里。所谓登录态对在线接口来说通常是一组 Cookie 和几个请求头字段的组合服务端靠它识别「这是一个已登录的合法客户端」。import requests import time session requests.Session() # 以下请求参数从本地会话导出仅供自己有权限的群使用 session.headers.update({ User-Agent: Mozilla/5.0, # 保持与导出会话时一致 Referer: fhttps://{host}/, # host 取自会话初始化返回 Cookie: config[cookie_str], # 导出的登录态 }) params { group_id: 10001, # 目标群号 start_index: 0, # 起始位置用于翻页 count: 50, # 每页数量建议不超过 50 } resp session.get(fhttps://{host}/cgi-bin/group_member, paramsparams) data resp.json()代码里有两个关键参数start_index是翻页起点和页码不是一回事每一页结束后服务端会返回下一段的起始偏移直接传页码大概率会拿到重复或缺失的数据count是单页数量不建议贪多单页拉太多反而容易被识别为异常行为50 是一个相对稳妥的取值。在线接口方案适合做增量不适合做全量。原因有两个一是每个群、每一页的请求都算一次在线访问量一大必定触发限流二是接口字段中有一部分是脱敏后的占位值拿不到真实 UIN这对需要精确用户标识的场景是硬伤。所以我的习惯是全量走本地增量走在线两条腿走路。3. 按数据层级把「群」拆成三个抓取对象3.1 群列表与群资料先摸清自己手里有哪些群拿到登录态之后第一件事不是直接进群抓成员而是先拉取自己的群列表。群列表是整个采集任务的地图——只有知道有哪些群、每个群的群号是什么后续按群抓取才有目标。群列表接口的分页参数比较特殊它不用页码而是用「起始位置 每页数量」的偏移方式。每次请求返回的 JSON 里除了当前页的群数据还会带一个是否到底的标记循环判断这个标记就能翻完所有群。def fetch_group_list(session, base_params, page_size30): all_groups [] start 0 while True: params dict(base_params) params[start_index] start params[count] page_size data session.get(fhttps://{host}/cgi-bin/group_list, paramsparams).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(fhttps://{host}/cgi-bin/group_member, paramsparams).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, limit50): msgs [] seq start_seq while True: params { group_id: group_id, start_seq: seq, # 从这条消息开始向前翻 count: limit, } data session.get(fhttps://{host}/cgi-bin/group_msg, paramsparams).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, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \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_asciiFalse是中文场景的必须项不加的话所有中文会变成\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, encodingutf-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_seqcheckpoint.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_days30): 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 条不是固定的要根据群规模调整——几百人的大群和几十人的小群标准必然不一样。把阈值参数暴露成配置项比写死在代码里实用得多。最后说一下我从这个项目里学到的习惯任何批量抓取任务都要有两套开关——总请求数上限和总运行时长上限到点自动停止不依赖人盯着。有一次我忘了加时长限制晚上挂机跑增量半夜接口异常返回了错误数据程序进入死循环刷了几万次请求第二天账号又进了安全保护名单。自那以后所有任务脚本第一行就是设置停止条件跑偏了也比跑疯了强。数据管道不是一次搭完就结束的东西每次客户端更新、接口调整都要重新核对字段和游标逻辑。保持增量、保持克制、保持对风控的敬畏这套系统就能长期稳定地给你供数。希望帮到你。本文还有配套的精品资源点击获取