影刀RPA RSS订阅自动化：信息源聚合与精选推送

影刀RPA RSS订阅自动化信息源聚合与精选推送作者林焱 | 适用人群影刀RPA新手 | 难度★★☆☆☆一、什么情况用这个每天的信息获取渠道太多了——技术博客、行业资讯、竞品动态、政策更新。一个个打开网站看一天什么都不用干了。用RSS订阅工具吧几十个订阅源混在一起真正有价值的内容淹没在信息洪流里。用影刀RPA定时拉取RSS订阅源 → 按关键词筛选 → 去重 → 生成精选日报 → 推送到微信/钉钉/邮件。每天只需要花5分钟看精选列表机器人帮你过滤掉90%的噪声。实际场景行业资讯监控、竞品动态追踪、技术博客聚合、政策法规更新提醒。这篇文章能解决的问题Python解析RSS/Atom订阅源按关键词自动筛选文章多源去重合并生成每日精选推送二、怎么做2.1 环境准备影刀安装目录\python\python.exe-mpipinstallfeedparser店群矩阵自动化突破运营极限2.2 影刀主流程┌──────────────────────────────────────────────┐ │ ① 【执行Python代码】 │ │ 拉取所有RSS源解析文章列表 │ │ ↓ │ │ ② 【执行Python代码】 │ │ 关键词筛选 去重 │ │ ↓ │ │ ③ 【执行Python代码】 │  │ 生成Markdown日报 │ │ ↓ │ │ ④ 【发送企业微信消息】 │ │ 推送精选日报 │ └──────────────────────────────────────────────┘2.3 核心代码步骤①拉取RSS源importfeedparserimportjsonfromdatetimeimportdatetime,timedelta# 你的RSS订阅源列表 RSS_SOURCES[# 技术博客{name:阮一峰的网络日志,url:https://www.ruanyifeng.com/blog/atom.xml,category:技术},{name:美团技术团队,url:https://tech.meituan.com/feed.xml,category:技术},# 行业资讯{name:36氪,url:https://36kr.com/feed,category:行业},{name:虎嗅,url:https://www.huxiu.com/rss/0.xml,category:行业},# 政策法规如果有RSS源的话{name:工信部,url:https://www.miit.gov.cn/rss,category:政策},]# 关键词配置 INCLUDE_KEYWORDS[RPA,自动化,人工智能,AI,低代码,数字化转型,影刀,机器人流程,大模型,LLM,]EXCLUDE_KEYWORDS[广告,推广,招聘,融资,# 过滤掉融资新闻和招聘广告]deffetch_articles(sources,days1):拉取所有RSS源的文章all_articles[]cutoffdatetime.now()-timedelta(daysdays)forsourceinsources:try:feedfeedparser.parse(source[url])# 检查是否解析成功iffeed.bozoandnotfeed.entries:print(f⚠️ 解析失败{source[name]}→{source[url]})continueforentryinfeed.entries:# 获取发布时间publishedNoneifhasattr(entry,published_parsed)andentry.published_parsed:publisheddatetime(*entry.published_parsed[:6])elifhasattr(entry,updated_parsed)andentry.updated_parsed:publisheddatetime(*entry.updated_parsed[:6])# 跳过过旧的文章ifpublishedandpublishedcutoff:continuearticle{source:source[name],category:source[category],title:entry.get(title,无标题),link:entry.get(link,),summary:entry.get(summary,),published:published.strftime(%Y-%m-%d %H:%M)ifpublishedelse未知,}all_articles.append(article)exceptExceptionase:print(f❌ 拉取失败{source[name]}→{e})print(f共拉取{len(all_articles)}篇文章)returnall_articles articlesfetch_articles(RSS_SOURCES,days1)outputjson.dumps(articles,ensure_asciiFalse)步骤②关键词筛选 去重importjsonimportrefromhtmlimportunescape articlesjson.loads(articles_data)# 从步骤①传入defclean_html(text):清洗HTML标签# 移除HTML标签textre.sub(r[^],,text)# 处理HTML实体textunescape(text)# 移除多余空白textre.sub(r\s, ,text).strip()returntextdefmatch_keywords(text,keywords):检查文本是否包含任一关键词text_lowertext.lower()forkwinkeywords:ifkw.lower()intext_lower:returnTruereturnFalse# 筛选 filtered[]seen_titlesset()# 用于去重forarticleinarticles:# 清洗摘要summaryclean_html(article[summary])# 关键词筛选full_textarticle[title] summaryifnotmatch_keywords(full_text,INCLUDE_KEYWORDS):continue# 排除关键词ifmatch_keywords(full_text,EXCLUDE_KEYWORDS):continue# 去重基于标题的前50个字符title_keyarticle[title][:50]iftitle_keyinseen_titles:continueseen_titles.add(title_key)article[summary_clean]summary[:200]# 摘要截断filtered.append(article)print(f筛选后剩余{len(filtered)}篇精选文章)filtered_outputjson.dumps(filtered,ensure_asciiFalse)步骤③生成日报importjson articlesjson.loads(filtered_articles)# 按分类分组fromcollectionsimportdefaultdict by_categorydefaultdict(list)forainarticles:by_category[a[category]].append(a)# 生成Markdown日报lines[]lines.append(f# 每日精选资讯 ({datetime.now().strftime(%Y-%m-%d)}))lines.append(f 共筛选{len(articles)}篇有价值文章)lines.append()forcategory,itemsinby_category.items():lines.append(f##{category})lines.append()fori,iteminenumerate(items,1):lines.append(f{i}. **[{item[title]}]({item[link]})**)lines.append(f - 来源{item[source]}|{item[published]})ifitem.get(summary_clean):lines.append(f -{item[summary_clean]})lines.append()report\n.join(lines)# 保存为文件可选# with open(fD:\\日报\\RSS日报_{datetime.now().strftime(%Y%m%d)}.md, w, encodingutf-8) as f:# f.write(report)print(f日报生成完成{len(articles)}篇)report_outputreport步骤④推送【发送企业微信消息】 类型Markdown 内容{report_output} ⚠️ 注意企微消息有长度限制4096字符如果日报太长需要分段发送。2.4 发送长文本超过限制时如果日报内容超过企业微信消息限制defsplit_markdown(content,max_length4000):将Markdown按二级标题拆分为多段sectionscontent.split(\n## )# 第一段保持不动result[sections[0]]forsectioninsections[1:]:section## section# 如果当前段加上新section不超过限制合并iflen(result[-1])len(section)max_length:result[-1]\nsectionelse:result.append(section)returnresult# 分段发送segmentssplit_markdown(report_output)fori,seginenumerate(segments):print(f发送第{i1}/{len(segments)}段...)# 在影刀中对每段执行【发送企业微信消息】三、有什么坑坑1有些网站不提供RSS很多网站已经停止维护RSS了。但好消息是可以用工具生成RSSRSSHub为几乎所有中文网站生成RSS开源项目可自建rss.app付费服务为无RSS的网站生成RSS示例通过RSSHub获取微博热搜https://rsshub.app/weibo/search/hot坑2feedparser处理不规范的RSS有些RSS源格式不规范缺少时间戳、编码错误、嵌套错误feedparser会标记bozo但尽力解析。temu店群自动化报活动案例feedfeedparser.parse(url)iffeed.bozo:print(f⚠️ RSS格式有问题但尝试继续解析)print(f异常类型{type(feed.bozo_exception).__name__})如果feed.entries为空才认为真的失败了。坑3关键词匹配的准确度简单字符串匹配会有假阳性——比如搜索AI会匹配到FAITH、MAIL这些无关词。改进方案用正则表达式做词边界匹配importredefmatch_keywords_smart(text,keywords):智能关键词匹配只匹配完整单词text_lowertext.lower()forkwinkeywords:# 中文字符直接搜索ifany(\u4e00c\u9fffforcinkw):ifkwintext_lower:returnTrueelse:# 英文单词用词边界匹配patternr\bre.escape(kw.lower())r\bifre.search(pattern,text_lower):returnTruereturnFalse坑4feedparser的编码问题有些中文RSS使用GBK编码feedparser可能解析乱码。# 如果标题/摘要乱码尝试指定编码feedfeedparser.parse(url)# feedparser通常能自动检测编码但如果不准iffeed.encodingandfeed.encoding.lower()!utf-8:# 手动重编码forentryinfeed.entries:entry.titleentry.title.encode(raw_unicode_escape).decode(utf-8,errorsignore)坑5RSS源的稳定性RSS源可能随时不可用服务器挂了、域名过期、格式变更。# 加上超时和重试importrequestsdeffetch_with_retry(url,retries3,timeout10):foriinrange(retries):try:responserequests.get(url,timeouttimeout,headers{User-Agent:YingdaoRPA/1.0})returnresponse.contentexceptrequests.exceptions.RequestExceptionase:ifiretries-1:time.sleep(2)else:raise# 用feedparser解析feedfeedparser.parse(fetch_with_retry(url))总结RSS 影刀 个人资讯秘书。feedparser解析RSS关键词做初筛去重避免重复阅读Markdown格式推送到聊天工具。核心价值是「过滤噪声」所以关键词要精心设计——太宽泛没过滤效果太严格可能漏掉重要信息。