用Python+Coze+飞书，我给自己做了个公众号AI日报机器人（附完整代码）
从零搭建个人AI资讯助手：Python+Coze+飞书全链路实战
每天打开微信，上百条未读红标让我头皮发麻。作为一个关注了87个技术类公众号的开发者，我发现自己90%的时间都在机械地滑动屏幕，却记不住任何有价值的内容。直到上个月，我用三个晚上时间搭建了一套完全属于自己的AI资讯过滤系统——它每天自动抓取我关注的公众号更新，通过大模型提炼核心观点，最终在飞书上生成一份简洁的日报。现在，我终于能每天用5分钟掌握真正需要的信息，而不是被算法投喂的碎片淹没。
1. 系统架构设计
这个私人资讯助手的核心由三个模块组成：微信爬虫负责内容采集，Coze平台进行智能处理，飞书机器人实现消息推送。整个系统的数据流向如下图所示：
[微信公众号] → [Python爬虫] → [Coze工作流] → [飞书机器人] → [用户]技术选型对比表：
| 组件 | 候选方案 | 最终选择理由 | 资源消耗 |
|---|---|---|---|
| 内容采集 | 影刀RPA/AnyProxy | Selenium+Requests组合最轻量 | 免费 |
| 内容处理 | 自建LLM/第三方API | Coze免费额度足够日常使用 | 500点/日 |
| 消息推送 | 企业微信/Server酱 | 飞书API免费额度高且稳定 | 1万次/月 |
提示：整套系统完全基于免费资源搭建，但需要注意Coze的点数消耗和飞书的API调用频率限制
2. 微信爬虫开发实录
2.1 模拟登录关键步骤
微信公众号后台的防爬机制主要依靠动态token和登录态验证。经过实测，最稳定的方案是组合使用Selenium模拟登录和Requests保持会话：
from selenium import webdriver from selenium.webdriver.common.by import By def wechat_login(): driver = webdriver.Chrome() driver.get("https://mp.weixin.qq.com") driver.find_element(By.XPATH, "//a[contains(@class,'login__type__container')]").click() # 等待用户扫码 while 'cgi-bin/home' not in driver.current_url: time.sleep(1) cookies = driver.get_cookies() driver.quit() return json.dumps({c['name']:c['value'] for c in cookies})避坑指南：
- 使用
undetected-chromedriver避免被识别为自动化工具 - 登录后获取的cookie有效期约72小时，需要定期刷新
- 推荐将cookie存储在环境变量中而非代码里
2.2 文章列表抓取优化
通过分析微信接口，我们发现获取文章列表需要两个关键参数：fakeid和token。以下是经过优化的采集函数：
def get_articles(cookies, accounts): session = requests.Session() session.cookies.update(json.loads(cookies)) # 获取动态token home_res = session.get("https://mp.weixin.qq.com", allow_redirects=False) token = re.search(r'token=(\d+)', home_res.headers['Location']).group(1) articles = [] for name in accounts: # 获取公众号fakeid params = { 'action': 'search_biz', 'token': token, 'query': name, 'count': 1 } resp = session.get("https://mp.weixin.qq.com/cgi-bin/searchbiz", params=params) fakeid = resp.json()['list'][0]['fakeid'] # 获取最新5篇文章 params = { 'action': 'list_ex', 'fakeid': fakeid, 'count': 5, 'token': token } resp = session.get("https://mp.weixin.qq.com/cgi-bin/appmsg", params=params) articles.extend(resp.json().get('app_msg_list', [])) return articles3. Coze工作流深度配置
3.1 工作流设计思路
在Coze平台创建的工作流主要完成三个任务：
- 单篇文章摘要提取（保留核心论点和数据）
- 多篇文章主题聚类（识别当日热点话题）
- 内容质量评分（过滤营销软文）
工作流节点配置示例：
[输入] → [文本清洗] → [摘要生成] → [主题分类] → [质量评分] → [汇总输出]3.2 API调用实战
Coze提供了完善的REST API，以下是Python调用示例：
def summarize_articles(articles, coze_token): url = "https://api.coze.cn/v1/workflow/run" headers = { "Authorization": f"Bearer {coze_token}", "Content-Type": "application/json" } summaries = [] for article in articles: data = { "workflow_id": "your_workflow_id", "parameters": { "url": article['link'], "min_length": 200 } } resp = requests.post(url, headers=headers, json=data) summaries.append(resp.json()['output']) return summaries注意：Coze免费账户每天有500点额度限制，单次调用通常消耗20-30点
4. 飞书机器人集成方案
4.1 机器人创建流程
- 登录飞书开放平台创建自建应用
- 在"权限管理"中添加
发送消息权限 - 在"事件订阅"中启用
接收消息v2.0 - 获取App ID和App Secret用于鉴权
4.2 消息推送实现
飞书支持多种消息类型，我们选择交互性更强的卡片消息：
def send_feishu_message(content, app_id, app_secret): # 获取access_token token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" token_resp = requests.post(token_url, json={ "app_id": app_id, "app_secret": app_secret }) access_token = token_resp.json()['tenant_access_token'] # 构建消息卡片 card = { "msg_type": "interactive", "card": { "elements": [{ "tag": "div", "text": {"content": content, "tag": "lark_md"} }], "header": { "title": {"content": "📰 每日AI资讯简报", "tag": "plain_text"} } } } # 发送消息 message_url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id" headers = {"Authorization": f"Bearer {access_token}"} requests.post(message_url, headers=headers, json=card)5. 系统部署与优化
5.1 定时任务管理
推荐使用APScheduler实现跨平台定时任务：
from apscheduler.schedulers.blocking import BlockingScheduler def daily_job(): cookies = refresh_cookies() articles = get_articles(cookies, ACCOUNTS) summaries = summarize_articles(articles, COZE_TOKEN) send_feishu_message(format_report(summaries), APP_ID, APP_SECRET) scheduler = BlockingScheduler() scheduler.add_job(daily_job, 'cron', hour=8, minute=30) scheduler.start()5.2 性能优化技巧
- 增量采集：记录已处理文章的URL避免重复分析
- 失败重试：对Coze API调用添加指数退避重试机制
- 本地缓存：使用SQLite存储临时数据防止中断丢失
- 流量控制：在免费额度内合理分配API调用次数
# 带重试机制的Coze调用 from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def safe_coze_call(params): return summarize_articles(params)现在每天早上8:30，我的飞书都会准时收到这样一份简报："今日AI领域3篇核心文章：1) Transformer架构新优化方案... 2) LangChain最佳实践... 3) 大模型微调技巧..."。整个系统跑在我的树莓派上，已经稳定运行了47天，帮我节省了至少100小时的无效阅读时间。