ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Python+清博API:构建带用户体系的公众号数据中台

Python+清博API:构建带用户体系的公众号数据中台 简介一套基于Python的微信公众号数据分析系统源码主要面向需要对公众号运营数据进行批量获取与统计的开发者、数据分析师或运营人员。项目通过调用清博大数据API实现文章标题、摘要、阅读量、点赞量等信息的抓取支持按关键词、公众号名称和日期范围搜索并覆盖10个月历史数据查询。同时提供单个公众号最新排名、文章总数、阅读总量、点赞总量及预估粉丝数等统计功能还包含用户登录验证与微信分组管理模块适合作为公众号数据采集与分析项目的学习模板或二次开发基础。压缩包共1055个文件约1.3MB其中以1032个class编译文件为主推测为项目依赖库或工具类另有6个Python脚本承载核心逻辑并包含少量js、xml、properties等配置资源便于理解项目结构。目前已有332人学习下载。透过这套源码可掌握API对接、数据封装、统计分析、分组管理等完整实现思路对想快速搭建公众号数据看板或研究Python后端项目组织的读者具有直接参考价值。1. 这不是爬虫是一个带用户体系的公众号数据中台运营同学隔三差五找你要“某某公众号最近涨了多少阅读”每次你都得打开清博、手动复制、再贴进Excel烦不烦这个基于Python的微信公众号数据分析系统就是把这件事从“人肉操作”变成“平台化查询”。它本身不爬公众号而是通过清博大数据开放API拿文章标题、摘要、阅读量、点赞量再落到自己的数据库里附带用户登录验证和分组管理。整套东西拿到手是一个带用户体系、分组管理、10个月历史数据查询的完整项目骨架适合正在做Python课设、需要公众号数据做分析、或者想给团队搭一个内部数据查询后台的人。它不是给你一个写死的脚本而是一个可以二次开发的半成品基座往下看怎么拆。2. 数据从哪来API接入、签名机制与数据表设计2.1 为什么用第三方API而不是自己写爬虫很多人一听到“微信公众号数据分析”就想到爬虫这其实是个误区。公众号文章页有反爬策略文章列表接口的签名算法隔三差五变更麻烦的是你抓到的是HTML页面还得自己解析正文里的JSON折腾一周可能连历史文章都翻不全。清博大数据这类平台的核心业务就是监测公众号数据它开放了标准API你传公众号名称和日期范围它直接返回结构化的阅读量和点赞量。这个项目走的就是这条路——调用清博API把数据落到本地数据库再做查询和统计。选第三方API还有个现实原因合规风险。公众号文章的阅读量是平台内部业务数据虽然有公开页面但大规模抓取始终是灰色地带。API是平台主动开放的有授权机制、有调用频率限制至少从源头避开了“被抓了别找我”这种坑。清博的API是按次计费的但这套代码的核心是演示“怎么规范地调用和存储”你把请求层换成别的数据源业务逻辑照样跑。2.2 请求签名先搞懂app_key和token的关系打开项目的main.py或者api_client.py核心逻辑其实就是一个带签名的HTTP请求。清博的接口鉴权方式很简单每个请求都必须带上app_key和一个动态生成的token。这个token不是你登录后拿到的那个而是每次请求前用app_key和当前时间戳拼接后做MD5算出来的。看代码import hashlib import time import requests APP_KEY your_app_key # 清博开放平台申请 SECRET your_secret # 你的密钥 def generate_token(): 清博接口签名app_key 时间戳 密钥拼接后做MD5 timestamp str(int(time.time())) raw_str APP_KEY timestamp SECRET token hashlib.md5(raw_str.encode(utf-8)).hexdigest() return token, timestamp def fetch_article_data(keyword, start_date, end_date, page1, limit20): 按关键词和时间范围拉取公众号文章列表 token, timestamp generate_token() params { app_key: APP_KEY, token: token, timestamp: timestamp, keyword: keyword, start_date: start_date, end_date: end_date, page: page, limit: limit } url https://api.gsdata.cn/api/v2/article/search resp requests.get(url, paramsparams, timeout10) return resp.json()这段代码的逻辑是把app_key、当前时间戳、SECRET拼成一个字符串做MD5得到token再把这个token随着请求一起发出去。服务器那边用同样的方式算一遍如果一致就认为是合法请求。参数里最容易被忽略的是timestamp很多接口会校验时间戳和服务器时间差超过5分钟就不认了所以这里用int(time.time())生成的是秒级时间戳别用什么毫秒级不然两边对不上。2.3 数据落库三张表撑起一个后台拿到JSON不能直接丢给前端你得存起来。项目里没有用ORM框架直接就是标准库sqlite3这对新手反而是件好事——少一层魔法看代码就能懂数据怎么流转。核心三张表articles存文章数据、users存登录账号、group_relation存分组关联关系。建表SQL长这样CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, summary TEXT, read_count INTEGER DEFAULT 0, like_count INTEGER DEFAULT 0, article_url TEXT UNIQUE, source_name TEXT, publish_date TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS group_relation ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_name TEXT NOT NULL, article_id INTEGER NOT NULL, FOREIGN KEY (article_id) REFERENCES articles(id) ON DELETE CASCADE );设计上有个小细节articles表的article_url加了UNIQUE约束这是防止重复插入的“后悔药”。清博API翻页时可能存在数据重叠同一篇文章在不同关键词下被搜出来两次如果你直接INSERT第二次就报错。我在做的时候会在插入前先做一次SELECT判断SELECT id FROM articles WHERE article_url ?存在就UPDATE阅读量不存在才INSERT。对10个月的数据量来说这种边查边插的方式效率完全够比纠结upsert语法来得实在。3. 用户与分组管理登录逻辑、会话保持与权限隔离3.1 登录验证的完整闭环这个项目不是简单的数据展示它带了用户系统所以整体框架是“登录才能查数据”。登录接口的逻辑不复杂前端传username和password后端接住后先去查users表比对password_hash字段。明文密码就别想了至少也要加个盐做MD5。项目里有一个典型的处理方式可以先看代码import hashlib import secrets from flask import Flask, request, session, jsonify app Flask(__name__) app.secret_key secrets.token_hex(16) # 每次启动随机生成生产环境必须固定 def verify_login(username, password): 验证用户名密码返回用户信息或None conn sqlite3.connect(wechat_data.db) cur conn.cursor() cur.execute(SELECT id, username, password_hash FROM users WHERE username ?, (username,)) row cur.fetchone() conn.close() if row is None: return None user_id, db_username, db_hash row # 前端传过来的密码拼接固定盐再做MD5 input_hash hashlib.md5((password fixed_salt).encode(utf-8)).hexdigest() if input_hash db_hash: return {id: user_id, username: db_username} return None这里的关键是salt不能每个用户随机生成否则你存salt的字段又多一个查起来麻烦。项目里是用一个固定的salt拼接密码做MD5安全性谈不上顶级但至少不是明文。真正的生产环境我建议直接用werkzeug自带的generate_password_hash它是加盐哈希安全性比手写MD5高一个量级。登录成功后用Flask的session存user_id后续请求通过session判断是否登录这比每次查一遍数据库要快得多。3.2 分组管理的幂等性设计分组功能是给公众号打标签用的比如“科技号”“生活号”“本地号”方便后面按组拉数据。项目里分组增删是两个接口但写的时候有个坑必须处理重复添加同名分组。前端不小心点两下提交按钮或者用户手滑创建了“科技号”和“科技号”后端不做判断数据库里就出现两条一样的group_name记录后续查数据就乱了。解决方式是插入前先查def add_group(group_name, article_id): 新增分组重复组名直接返回已存在 conn sqlite3.connect(wechat_data.db) cur conn.cursor() cur.execute(SELECT id FROM group_relation WHERE group_name ? AND article_id ?, (group_name, article_id)) exists cur.fetchone() if exists: conn.close() return {code: 1, msg: 该分组的关联关系已存在} cur.execute(INSERT INTO group_relation (group_name, article_id) VALUES (?, ?), (group_name, article_id)) conn.commit() conn.close() return {code: 0, msg: 添加成功}注意这里的幂等不是全局唯一而是“同一个组名和同一篇文章的关联关系唯一”。同一个组名下可以绑多篇文章同一篇文章也可以出现在不同组里这是典型的多对多关系。删除分组同理先检查article_id有没有被别的表引用直接DELETE会造成外键悬空所以建表时写了ON DELETE CASCADE删文章时关联记录自动清掉这就不会出现幽灵关联了。3.3 代码组织参考项目的utils拆分方式下载下来的源码包里目录结构能看出来作者是按功能拆分的有单独的api模块放请求封装有db模块放数据库操作views模块放路由处理。这样的布局对二次开发很友好你要换数据源就只改api层要加字段只动db层不动路由。很多人拿到项目第一步就改核心逻辑这是错的做法。我先花半小时把每个.py文件里的def列表打出来看一遍函数名大致知道哪个文件管哪摊事再动手改代码。改的时候养个习惯每个文件只做一件事api_client.py里别写SQL语句db.py里别出现requests调用这样出问题才好排查。4. 避坑手册从下载到跑通的五个真实翻车现场4.1 解压后看到一堆.class文件怀疑拿错包现象解压zip之后看到的是一堆Lang.class、NutDao.class、ClassWriter.class这种文件作为Python项目却出现一堆Java编译产物第一反应是发错资源了。原因这套代码的依赖包里混进了第三方Java库就是项目文件列表里那些nutzboot框架的编译类可能是打包时误把整个依赖目录塞了进来不影响Python主体代码。解决忽略这些.class文件直接找.py文件。正常情况下压缩包里有main.py、api_client.py、db_helper.py这些核心文件如果有完整的requirements.txt或说明文档按那个跑即可。我吃到这个教训后下载完源码第一件事不是解压而是看文件清单——凡是跟项目技术栈不符的文件直接跳过省得被干扰。4.2 接口返回401但app_key确实没填错现象照着文档填了APP_KEY和SECRET请求发出去清博返回401 Unauthorized。原因八成是token生成方式和服务器不一致。清博的签名要求是app_key、timestamp、token三段都传但token的计算顺序有人会写反比如先拼SECRET再拼时间戳。还有一种是时间戳的单位问题Java那边很多用毫秒Python这边用秒两边服务器校验对不上。解决打开官方文档对着签名规则逐字核对拼接顺序另外检查一下服务器时间是否和本地偏差太大我遇到过本机时钟慢了3分钟导致签名全部失败的同步一下系统时间就好了。4.3 阅读量为0但公众号明显有点击现象拉某篇热门文章的接口数据read_count返回的是0但文章实际阅读过万。原因清博对部分公众号平台做了数据脱敏小号或者新号的文章可能只有“--”这种占位符接口解析时转成int失败默认给了0。项目的json解析代码里如果直接int(data[read_count])遇到空字符串或“--”就会报错被try-except接住后赋了0。解决写个防御性转换函数遇到非数字字符串一律返回NoneNone不参与统计。这件事教会我第三方API返回的字段值永远不要假设它是规范数字解析前先打印原始JSON看一眼比啥都好使。4.4 查询范围超过10个月直接报错现象用代码拉一年前的文章接口返回异常或数据为空。原因清博API对非付费用户开放的历史数据窗口就是10个月这是平台限制不是代码bug。项目摘要里也写了“提供10个月的历史数据查询”说明作者默认了你在10个月窗口内操作。解决代码里加一层日期范围校验start_date和end_date的差值超过300天就直接提示“超出可查询范围”把前端查询框的日历控件也限制在10个月。别想着绕过这个限制调用第三方API这种黑匣子平台说不给的数据就是不给绕来绕去最后被封了账号更亏。4.5 删除分组的接口一直报500现象前端调删除分组接口后端日志里看到IntegrityError500崩了。原因group_relation表里article_id设了外键删除分组时如果先删group_relation再删articles顺序反了就会外键冲突。项目里如果先执行了DELETE FROM articles WHERE article_id xxx再把关联的group_relation删掉数据库直接拒绝。解决调整删除顺序先删关联表记录再删文章或者干脆在SQLite里开启外键约束PRAGMA foreign_keys ON;让级联删除接管。从那以后我写删除逻辑都有个条件反射凡是涉及外键的SQL先问自己是“谁引用谁”再决定DELETE的先后顺序。5. 拿到数据之后导出Excel和交叉校验的两个实用技巧数据都进库了你肯定想把它导出来发给运营看。项目自带的输出格式比较简陋无非是控制台打印或者简单HTML表格。实际工作中我更常用的是直接给运营发CSV他们就能拖进Excel里做透视表。这里有个小坑Python的csv模块默认用逗号分隔但中文内容里经常自带逗号一导Excel就串列。我一般会换成tab分隔再给文件名加上日期import csv def export_to_csv(articles, filename): 导出文章数据为TSV文件避免中文逗号导致列错位 with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.writer(f, delimiter\t) writer.writerow([标题, 摘要, 阅读量, 点赞量, 发布日期]) for item in articles: writer.writerow([ item[title], item.get(summary, ), item[read_count], item[like_count], item[publish_date] ])编码这里我吃过亏用utf-8导出的CSV用Excel打开显示乱码因为Excel默认按GBK读文件。后来统一加utf-8-sig编码也就是在文件开头写入BOM标记Excel就能正确识别。这算是我最常用的一招谁用谁知道。另一个技巧是交叉校验数据准确性当拉取到的单篇文章阅读量异常高或异常低时去清博的网页端手动搜一下该公众号看数值是否对得上。我一般会写一个抽查脚本随机取每天数据里的5篇文章拿接口返回的阅读量跟清博前台显示的数值做比对误差超过1%就说明当天可能有清洗逻辑出问题。这套校验做完导出给别人的数据自己心里才有底。从那以后我每次对接第三方API都强制自己先做数据交叉验证哪怕多花半小时也不能把没校准过的数据交出去。希望帮到你。本文还有配套的精品资源点击获取
返回列表