1. 从手动刷榜到自动推送这个任务到底在解决什么问题每天早上打开电脑第一件事就是刷一圈AI产品榜单看看有没有新东西冒出来——这个动作我持续了大半年。刚开始还挺新鲜时间一长就发现这件事本质上跟以前刷RSS、刷论坛没什么区别信息是散的注意力是碎的而且你永远不知道自己是不是漏掉了什么。更麻烦的是团队里其他人也在做同样的事大家各自刷各自的信息不共享等到开会聊起来才发现有人三天前就看到的东西另一个人今天才刷到。这个项目的出发点特别朴素能不能让机器替我把这件事干了每天早上自动跑一遍把结果整理好推给我。标题里说的“Hermes”是我给这套自动化流程起的一个代号你可以把它理解成一个跑在后台的定时任务调度器加上一套信息采集和推送逻辑。它不复杂但确实把我从每天二十分钟的重复劳动里解放出来了。具体来说这套东西要解决三个层面的问题。第一层是信息采集的自动化不用手动去各个平台翻而是让程序按预设的规则去抓取和筛选。第二层是信息处理的标准化不同来源的数据格式不一样有的给标题有的给简介有的只有链接需要统一成一套结构化的格式方便后续处理。第三层是推送的定时化和渠道化不是随时推而是固定时间推不是推到一个地方而是推到你实际会看的地方。适合谁来参考这套方案如果你符合下面任意一条这篇文章应该对你有用你每天需要花时间跟踪某个垂直领域的新动态你所在的团队有信息同步的需求但一直没找到好办法你对定时任务、数据抓取、消息推送这些技术点有兴趣想找一个完整的练手项目。不需要你是资深工程师但至少要能看懂基本的脚本逻辑知道什么是API、什么是JSON不然实操部分会比较吃力。我先把整体架构说清楚再往下拆细节。整个流程分四段触发层负责定时唤醒任务采集层负责去各个源头拿数据处理层负责清洗、去重、排序推送层负责把最终结果送到你面前。每一层都有不同的技术选型和实现方式我会把选型理由和踩过的坑都讲清楚。2. 整体架构设计与技术选型思路2.1 为什么选定时任务而不是实时监听一开始我想过做实时监听就是那种一有更新就立刻推的模式。后来放弃了原因有两个。第一AI产品这个领域的信息更新频率没有高到需要实时响应的程度一天集中处理一次完全够用实时监听反而会造成信息碎片化一条一条推过来注意力被反复打断。第二实时监听对资源的要求更高需要维持长连接或者频繁轮询对于一个个人项目来说性价比不高。定时任务的好处是节奏可控。我设定的是每天早上八点跑一次跑完之后把结果整理成一份摘要八点半推送到位。这样我到了工位打开手机就能看到不需要自己去触发任何操作。而且定时任务的调试成本低跑失败了可以手动重跑不会影响其他环节。注意定时任务的时间设定要避开目标数据源的高峰期。有些平台在早晚高峰时段响应会变慢如果你刚好在那个时间段去抓取失败率会明显上升。我的经验是早上六点到七点之间跑采集这个时候大部分服务都比较空闲。2.2 采集层多源聚合还是单源深挖采集层的设计取决于你的信息需求。如果你只关心某一个特定平台的数据那单源采集就够了逻辑简单维护成本低。但实际情况下单一来源往往覆盖不全你需要从多个地方拿数据才能拼出完整的图景。我采用的是多源聚合的方案同时从三个类型的源头采集产品榜单类、社区讨论类、官方发布类。每一类的采集策略不一样。榜单类通常是结构化的列表页解析相对容易社区讨论类需要从帖子标题和摘要里提取关键信息噪音比较大官方发布类更新频率低但信息密度高需要重点标记。多源聚合带来的最大问题是数据格式不统一。有的源返回的是JSON有的是HTML需要解析有的甚至只有纯文本。我的处理方式是在采集层和处 理层之间加一个适配器层每个源对应一个适配器负责把原始数据转换成统一的内部格式。这样后面处理层就不需要关心数据是从哪来的只管处理标准格式就行。2.3 处理层去重、排序、打标一个都不能少采集回来的原始数据不能直接推里面会有大量重复和低质量内容。处理层要做三件事。去重是第一件。不同源之间会有重叠同一个产品可能在榜单里出现一次在社区讨论里又出现一次。我的去重策略是基于URL和标题的双重指纹先看URL是否相同如果URL不同但标题相似度超过阈值也判定为重复。相似度计算用的是简单的编辑距离没有上复杂的模型因为在这个场景下够用了。排序是第二件。推送给你的列表不能是乱序的需要有优先级。我的排序规则是官方发布类排最前因为信息最权威其次是榜单类中新上榜或者排名变化大的最后是社区讨论类中热度高的。热度用回复数和点赞数加权计算权重系数根据实际效果调过几次。打标是第三件。每条信息我会打上几个标签比如“新发布”“排名上升”“讨论热烈”“可能相关”。标签的作用是让你在快速浏览时能一眼抓住重点不需要逐条读全文。2.4 推送层选对渠道比选对工具更重要推送渠道的选择标准只有一个你实际会看的地方。我试过邮件、即时通讯工具、甚至短信最后发现最有效的还是推送到日常使用频率最高的那个通讯工具上。邮件的问题是打开率低短信的问题是信息太短说不清楚即时通讯工具刚好在中间既能承载富文本又能保证你看到。推送格式我调整过好几版。最早是纯文本列表后来改成带链接的Markdown格式再后来加了摘要和标签。现在的格式是标题加链接加一句话摘要加标签每条控制在三行以内整个推送不超过十五条。超过十五条的会折叠成“更多内容请查看完整报告”完整报告存在本地或者云端需要的时候再去翻。提示推送频率不要太高。我一开始设了早中晚三次后来发现中午和晚上那两次根本没人看反而造成了信息过载。现在只保留早上一次效果最好。3. 核心环节实操从零搭建一套可运行的流程3.1 环境准备与依赖安装这套流程跑在Linux环境下用Python作为主要开发语言。选Python的原因很简单生态成熟各种采集、处理、推送的库都有现成的不需要自己造轮子。基础环境需要Python 3.8以上pip包管理工具以及一个可以跑定时任务的系统。核心依赖库包括requests用于HTTP请求beautifulsoup4用于HTML解析schedule用于定时任务调度python-dateutil用于时间处理。如果你打算把结果存到数据库还需要sqlite3或者pymysql。推送部分取决于你选的渠道如果用即时通讯工具的机器人接口通常只需要requests就够了。pip install requests beautifulsoup4 schedule python-dateutil安装完之后先跑一个简单的测试脚本确认网络请求和解析都正常。这一步看起来多余但实际上能帮你提前发现环境问题比如SSL证书错误、代理配置问题等。我当初就是跳过了这一步结果后面调试采集逻辑的时候花了半天时间才发现是环境问题。3.2 采集适配器的编写要点每个采集适配器本质上就是一个函数输入是源地址和参数输出是统一格式的列表。统一格式我定义成这样的字典结构{ title: 产品名称或帖子标题, url: 原始链接, summary: 一句话摘要, source: 来源标识, timestamp: 发布时间戳, raw: {} # 原始数据备用 }写适配器的时候有几个坑要注意。第一是请求头很多站点会检查User-Agent如果不设置成常见的浏览器标识请求会被拒绝。第二是频率控制不要在短时间内对同一个源发大量请求我一般会在两次请求之间加一到两秒的延迟。第三是异常处理网络请求失败是常态要有重试机制但重试次数不要超过三次否则会卡住整个流程。import requests import time from bs4 import BeautifulSoup def fetch_source_a(url): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } for attempt in range(3): try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() return parse_source_a(resp.text) except Exception as e: if attempt 2: raise time.sleep(2)解析部分根据源的类型不同而不同。如果是JSON接口直接用resp.json()拿数据如果是HTML页面用BeautifulSoup定位元素。定位元素的时候尽量用稳定的选择器不要依赖会变化的class名。我一般会优先找id属性或者>from difflib import SequenceMatcher def deduplicate(items): seen_urls set() seen_titles [] result [] for item in items: if item[url] in seen_urls: continue is_duplicate False for existing_title in seen_titles: ratio SequenceMatcher(None, item[title], existing_title).ratio() if ratio 0.85: is_duplicate True break if is_duplicate: continue seen_urls.add(item[url]) seen_titles.append(item[title]) result.append(item) return result相似度阈值设成0.85是试出来的。设太低会误杀把不同但相似的产品当成同一个设太高会漏掉明明是一个东西但标题写法不同就判成两个。0.85在我的数据集上表现比较均衡你可以根据自己的数据调。排序函数根据来源和热度计算一个分数然后按分数降序排列。来源权重我设的是官方发布1.0榜单0.8社区0.6。热度分数用回复数和点赞数加权回复数权重0.6点赞数权重0.4。这两个权重也是调出来的逻辑是回复比点赞更能反映讨论热度。3.4 推送消息的组装与发送推送消息的组装分两步。第一步是把处理后的列表转换成Markdown格式的文本第二步是调用推送渠道的接口发送。Markdown格式我固定成这样的模板今日AI产品动态{date} {for item in items} - [{title}]({url}) {summary} | 标签{tags} {end}发送部分取决于渠道。如果是即时通讯工具的机器人通常是一个POST请求把消息内容放在body里。注意有些渠道对消息长度有限制超过限制需要分段发送。我一般会在组装阶段就控制总长度超过阈值就截断并提示“完整内容请查看报告”。def send_notification(webhook_url, content): payload { msgtype: markdown, markdown: { content: content } } resp requests.post(webhook_url, jsonpayload, timeout10) return resp.status_code 200发送失败的处理策略是重试两次如果还是失败就记录日志不阻塞后续流程。推送失败不是致命问题第二天还会再跑但日志要留好方便排查是网络问题还是接口问题。3.5 定时任务的配置与监控定时任务用schedule库来实现逻辑很简单定义一个函数里面按顺序调用采集、处理、推送然后用schedule.every().day.at(08:00).do(job)注册最后跑一个无限循环。import schedule import time def job(): raw collect_all() processed process(raw) content format_message(processed) send_notification(webhook_url, content) schedule.every().day.at(08:00).do(job) while True: schedule.run_pending() time.sleep(60)这个方案适合跑在一直开机的机器上。如果你用的是云服务器建议用系统自带的定时任务工具来触发比Python层面的调度更稳定。监控方面我加了一个简单的日志记录每次跑完把结果数量、成功失败状态写到一个日志文件里。每周翻一次日志看看有没有异常。注意定时任务的时区要确认清楚。有些服务器默认是UTC时间如果你按本地时间设的八点实际跑起来可能是下午四点。我踩过这个坑后来在脚本里显式设置了时区。4. 实操中遇到的典型问题与排查方法4.1 采集失败从表象到根因的排查路径采集失败是最常见的问题表现是某个源返回空列表或者直接报错。排查的时候按这个顺序走先看网络通不通用curl或者ping测试一下目标地址再看请求头对不对有些站点会屏蔽默认的Python User-Agent然后看解析逻辑有没有问题可能是页面结构变了导致选择器失效。我遇到过一次比较隐蔽的情况请求返回200但内容是空的。查了半天才发现是目标站点做了反爬检测到频繁请求后返回了一个空页面。解决办法是降低请求频率并且在请求头里加上Referer字段模拟从站内跳转过来的正常访问。还有一个常见问题是编码。有些站点的响应不是UTF-8直接按UTF-8解析会乱码。处理方式是在请求之后先检测编码用resp.apparent_encoding来自动判断然后再设置resp.encoding。4.2 推送内容格式错乱的处理推送格式错乱通常是因为Markdown语法和渠道的解析规则冲突。比如有些渠道不支持嵌套列表你写了嵌套结构就会显示异常。解决办法是尽量用最简单的Markdown语法只用标题、加粗、链接和普通列表不要用表格和代码块。另一个问题是特殊字符转义。产品名称里如果有方括号或者竖线会破坏Markdown的链接语法。我加了一个转义函数把可能冲突的字符替换成全角或者加上反斜杠。def escape_markdown(text): for char in [[, ], |, *, _]: text text.replace(char, \\ char) return text4.3 去重逻辑的误杀与漏杀去重逻辑调参是个细致活。误杀是指把不同的东西判成重复漏杀是指重复的东西没被去掉。误杀的危害更大因为你会漏掉重要信息。我一开始把阈值设成0.7结果“AI写作助手”和“AI绘画助手”被判成重复直接漏掉了一个品类。后来调到0.85才正常。漏杀的情况也有比如同一个产品在不同源上的标题一个是中文一个是英文编辑距离算出来相似度很低就不会被判重。这种只能靠人工规则补充比如维护一个产品名称的别名表在去重之前先做一轮别名归一化。4.4 定时任务不执行的几种可能定时任务不执行先检查进程还在不在。如果是用schedule库跑的脚本进程挂了任务自然就不跑了。建议用systemd或者supervisor来管理进程挂了能自动拉起来。如果进程在但任务没跑检查时间设置。schedule库用的是本地时间但如果你在容器里跑容器的时区可能是UTC。还有一个可能是任务执行时间太长超过了间隔时间导致下一次触发被跳过。我的处理方式是加一个锁机制任务没跑完之前不允许再次触发。问题现象可能原因排查方法解决措施采集返回空反爬拦截检查请求头和频率降低频率加Referer推送格式乱Markdown冲突检查特殊字符转义或简化语法去重误杀阈值过低抽查判重结果调高阈值至0.85任务不执行进程挂掉或时区错误检查进程和时区用进程管理工具显式设时区4.5 几个让我少走弯路的实操心得第一个心得是先跑通再优化。我一开始想把所有功能都做完美再上线结果拖了两周还没跑起来。后来改变策略先用最粗糙的方式跑通全流程哪怕采集只有一个源、推送只是纯文本先让它跑起来然后再逐步替换和优化。这样每天都有反馈调整方向也更明确。第二个心得是日志比调试器好用。这种定时任务出问题的时候往往是半夜或者清晨你不可能守着调试。把关键步骤的输入输出都记到日志里出了问题翻日志比重新跑一遍快得多。日志级别我一般设成INFO关键错误用ERROR方便过滤。第三个心得是推送内容宁少勿多。我一开始想把所有采集到的都推过去结果每天三四十条根本看不过来。后来改成只推排序后的前十五条反而每条都会认真看。信息过载比信息不足更可怕因为过载会导致你直接忽略全部。第四个心得是定期回顾和调整。这套流程不是设好就一劳永逸的数据源会变你的关注点也会变。我每个月会花十分钟看一下推送记录把不再关注的源去掉把新发现的源加进来。保持流程和需求对齐比追求技术上的完美更重要。5. 扩展方向这套流程还能怎么用这套东西跑顺了之后我发现它的核心逻辑其实可以迁移到很多场景。把“AI产品”换成任何你关注的垂直领域采集源换一下处理规则调一下推送渠道不变就是一个新的信息跟踪工具。我后来用同样的架构做了一个行业政策更新的跟踪把采集源换成几个官方发布页面处理逻辑简化成只做去重和排序推送频率改成每周一次效果也很好。另一个扩展方向是增加分析维度。现在的处理层只做了去重、排序、打标还可以加趋势分析。比如记录每个产品每天的出现次数画一个热度变化曲线当某个产品的热度突然上升时自动标记为“异动”。这个功能需要把历史数据存下来用简单的时序分析就能做不需要复杂的模型。还有一个方向是多用户支持。现在的流程是单用户的推送内容按我的偏好来排序。如果团队里多个人用可以加一个偏好配置每个人设置自己关注的标签和来源推送的时候按人过滤。这个改动主要在推送层采集和处理层可以复用。最后说一个我实际用下来觉得最有价值的点这套流程最大的收益不是省了那二十分钟而是建立了一个稳定的信息输入节奏。以前是随机刷、随机看现在是固定时间、固定格式、固定数量。节奏稳定了你对信息的消化效率会明显提升不会出现“刷了很多但什么都没记住”的情况。这个价值比自动化本身更大。 
长河君 · 2026/10/20 22:10:13 · 阅读 1.2k 基于Python的定时任务自动化：AI产品信息采集与推送实践
1. 从手动刷榜到自动推送这个任务到底在解决什么问题每天早上打开电脑第一件事就是刷一圈AI产品榜单看看有没有新东西冒出来——这个动作我持续了大半年。刚开始还挺新鲜时间一长就发现这件事本质上跟以前刷RSS、刷论坛没什么区别信息是散的注意力是碎的而且你永远不知道自己是不是漏掉了什么。更麻烦的是团队里其他人也在做同样的事大家各自刷各自的信息不共享等到开会聊起来才发现有人三天前就看到的东西另一个人今天才刷到。这个项目的出发点特别朴素能不能让机器替我把这件事干了每天早上自动跑一遍把结果整理好推给我。标题里说的“Hermes”是我给这套自动化流程起的一个代号你可以把它理解成一个跑在后台的定时任务调度器加上一套信息采集和推送逻辑。它不复杂但确实把我从每天二十分钟的重复劳动里解放出来了。具体来说这套东西要解决三个层面的问题。第一层是信息采集的自动化不用手动去各个平台翻而是让程序按预设的规则去抓取和筛选。第二层是信息处理的标准化不同来源的数据格式不一样有的给标题有的给简介有的只有链接需要统一成一套结构化的格式方便后续处理。第三层是推送的定时化和渠道化不是随时推而是固定时间推不是推到一个地方而是推到你实际会看的地方。适合谁来参考这套方案如果你符合下面任意一条这篇文章应该对你有用你每天需要花时间跟踪某个垂直领域的新动态你所在的团队有信息同步的需求但一直没找到好办法你对定时任务、数据抓取、消息推送这些技术点有兴趣想找一个完整的练手项目。不需要你是资深工程师但至少要能看懂基本的脚本逻辑知道什么是API、什么是JSON不然实操部分会比较吃力。我先把整体架构说清楚再往下拆细节。整个流程分四段触发层负责定时唤醒任务采集层负责去各个源头拿数据处理层负责清洗、去重、排序推送层负责把最终结果送到你面前。每一层都有不同的技术选型和实现方式我会把选型理由和踩过的坑都讲清楚。2. 整体架构设计与技术选型思路2.1 为什么选定时任务而不是实时监听一开始我想过做实时监听就是那种一有更新就立刻推的模式。后来放弃了原因有两个。第一AI产品这个领域的信息更新频率没有高到需要实时响应的程度一天集中处理一次完全够用实时监听反而会造成信息碎片化一条一条推过来注意力被反复打断。第二实时监听对资源的要求更高需要维持长连接或者频繁轮询对于一个个人项目来说性价比不高。定时任务的好处是节奏可控。我设定的是每天早上八点跑一次跑完之后把结果整理成一份摘要八点半推送到位。这样我到了工位打开手机就能看到不需要自己去触发任何操作。而且定时任务的调试成本低跑失败了可以手动重跑不会影响其他环节。注意定时任务的时间设定要避开目标数据源的高峰期。有些平台在早晚高峰时段响应会变慢如果你刚好在那个时间段去抓取失败率会明显上升。我的经验是早上六点到七点之间跑采集这个时候大部分服务都比较空闲。2.2 采集层多源聚合还是单源深挖采集层的设计取决于你的信息需求。如果你只关心某一个特定平台的数据那单源采集就够了逻辑简单维护成本低。但实际情况下单一来源往往覆盖不全你需要从多个地方拿数据才能拼出完整的图景。我采用的是多源聚合的方案同时从三个类型的源头采集产品榜单类、社区讨论类、官方发布类。每一类的采集策略不一样。榜单类通常是结构化的列表页解析相对容易社区讨论类需要从帖子标题和摘要里提取关键信息噪音比较大官方发布类更新频率低但信息密度高需要重点标记。多源聚合带来的最大问题是数据格式不统一。有的源返回的是JSON有的是HTML需要解析有的甚至只有纯文本。我的处理方式是在采集层和处 理层之间加一个适配器层每个源对应一个适配器负责把原始数据转换成统一的内部格式。这样后面处理层就不需要关心数据是从哪来的只管处理标准格式就行。2.3 处理层去重、排序、打标一个都不能少采集回来的原始数据不能直接推里面会有大量重复和低质量内容。处理层要做三件事。去重是第一件。不同源之间会有重叠同一个产品可能在榜单里出现一次在社区讨论里又出现一次。我的去重策略是基于URL和标题的双重指纹先看URL是否相同如果URL不同但标题相似度超过阈值也判定为重复。相似度计算用的是简单的编辑距离没有上复杂的模型因为在这个场景下够用了。排序是第二件。推送给你的列表不能是乱序的需要有优先级。我的排序规则是官方发布类排最前因为信息最权威其次是榜单类中新上榜或者排名变化大的最后是社区讨论类中热度高的。热度用回复数和点赞数加权计算权重系数根据实际效果调过几次。打标是第三件。每条信息我会打上几个标签比如“新发布”“排名上升”“讨论热烈”“可能相关”。标签的作用是让你在快速浏览时能一眼抓住重点不需要逐条读全文。2.4 推送层选对渠道比选对工具更重要推送渠道的选择标准只有一个你实际会看的地方。我试过邮件、即时通讯工具、甚至短信最后发现最有效的还是推送到日常使用频率最高的那个通讯工具上。邮件的问题是打开率低短信的问题是信息太短说不清楚即时通讯工具刚好在中间既能承载富文本又能保证你看到。推送格式我调整过好几版。最早是纯文本列表后来改成带链接的Markdown格式再后来加了摘要和标签。现在的格式是标题加链接加一句话摘要加标签每条控制在三行以内整个推送不超过十五条。超过十五条的会折叠成“更多内容请查看完整报告”完整报告存在本地或者云端需要的时候再去翻。提示推送频率不要太高。我一开始设了早中晚三次后来发现中午和晚上那两次根本没人看反而造成了信息过载。现在只保留早上一次效果最好。3. 核心环节实操从零搭建一套可运行的流程3.1 环境准备与依赖安装这套流程跑在Linux环境下用Python作为主要开发语言。选Python的原因很简单生态成熟各种采集、处理、推送的库都有现成的不需要自己造轮子。基础环境需要Python 3.8以上pip包管理工具以及一个可以跑定时任务的系统。核心依赖库包括requests用于HTTP请求beautifulsoup4用于HTML解析schedule用于定时任务调度python-dateutil用于时间处理。如果你打算把结果存到数据库还需要sqlite3或者pymysql。推送部分取决于你选的渠道如果用即时通讯工具的机器人接口通常只需要requests就够了。pip install requests beautifulsoup4 schedule python-dateutil安装完之后先跑一个简单的测试脚本确认网络请求和解析都正常。这一步看起来多余但实际上能帮你提前发现环境问题比如SSL证书错误、代理配置问题等。我当初就是跳过了这一步结果后面调试采集逻辑的时候花了半天时间才发现是环境问题。3.2 采集适配器的编写要点每个采集适配器本质上就是一个函数输入是源地址和参数输出是统一格式的列表。统一格式我定义成这样的字典结构{ title: 产品名称或帖子标题, url: 原始链接, summary: 一句话摘要, source: 来源标识, timestamp: 发布时间戳, raw: {} # 原始数据备用 }写适配器的时候有几个坑要注意。第一是请求头很多站点会检查User-Agent如果不设置成常见的浏览器标识请求会被拒绝。第二是频率控制不要在短时间内对同一个源发大量请求我一般会在两次请求之间加一到两秒的延迟。第三是异常处理网络请求失败是常态要有重试机制但重试次数不要超过三次否则会卡住整个流程。import requests import time from bs4 import BeautifulSoup def fetch_source_a(url): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } for attempt in range(3): try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() return parse_source_a(resp.text) except Exception as e: if attempt 2: raise time.sleep(2)解析部分根据源的类型不同而不同。如果是JSON接口直接用resp.json()拿数据如果是HTML页面用BeautifulSoup定位元素。定位元素的时候尽量用稳定的选择器不要依赖会变化的class名。我一般会优先找id属性或者>from difflib import SequenceMatcher def deduplicate(items): seen_urls set() seen_titles [] result [] for item in items: if item[url] in seen_urls: continue is_duplicate False for existing_title in seen_titles: ratio SequenceMatcher(None, item[title], existing_title).ratio() if ratio 0.85: is_duplicate True break if is_duplicate: continue seen_urls.add(item[url]) seen_titles.append(item[title]) result.append(item) return result相似度阈值设成0.85是试出来的。设太低会误杀把不同但相似的产品当成同一个设太高会漏掉明明是一个东西但标题写法不同就判成两个。0.85在我的数据集上表现比较均衡你可以根据自己的数据调。排序函数根据来源和热度计算一个分数然后按分数降序排列。来源权重我设的是官方发布1.0榜单0.8社区0.6。热度分数用回复数和点赞数加权回复数权重0.6点赞数权重0.4。这两个权重也是调出来的逻辑是回复比点赞更能反映讨论热度。3.4 推送消息的组装与发送推送消息的组装分两步。第一步是把处理后的列表转换成Markdown格式的文本第二步是调用推送渠道的接口发送。Markdown格式我固定成这样的模板今日AI产品动态{date} {for item in items} - [{title}]({url}) {summary} | 标签{tags} {end}发送部分取决于渠道。如果是即时通讯工具的机器人通常是一个POST请求把消息内容放在body里。注意有些渠道对消息长度有限制超过限制需要分段发送。我一般会在组装阶段就控制总长度超过阈值就截断并提示“完整内容请查看报告”。def send_notification(webhook_url, content): payload { msgtype: markdown, markdown: { content: content } } resp requests.post(webhook_url, jsonpayload, timeout10) return resp.status_code 200发送失败的处理策略是重试两次如果还是失败就记录日志不阻塞后续流程。推送失败不是致命问题第二天还会再跑但日志要留好方便排查是网络问题还是接口问题。3.5 定时任务的配置与监控定时任务用schedule库来实现逻辑很简单定义一个函数里面按顺序调用采集、处理、推送然后用schedule.every().day.at(08:00).do(job)注册最后跑一个无限循环。import schedule import time def job(): raw collect_all() processed process(raw) content format_message(processed) send_notification(webhook_url, content) schedule.every().day.at(08:00).do(job) while True: schedule.run_pending() time.sleep(60)这个方案适合跑在一直开机的机器上。如果你用的是云服务器建议用系统自带的定时任务工具来触发比Python层面的调度更稳定。监控方面我加了一个简单的日志记录每次跑完把结果数量、成功失败状态写到一个日志文件里。每周翻一次日志看看有没有异常。注意定时任务的时区要确认清楚。有些服务器默认是UTC时间如果你按本地时间设的八点实际跑起来可能是下午四点。我踩过这个坑后来在脚本里显式设置了时区。4. 实操中遇到的典型问题与排查方法4.1 采集失败从表象到根因的排查路径采集失败是最常见的问题表现是某个源返回空列表或者直接报错。排查的时候按这个顺序走先看网络通不通用curl或者ping测试一下目标地址再看请求头对不对有些站点会屏蔽默认的Python User-Agent然后看解析逻辑有没有问题可能是页面结构变了导致选择器失效。我遇到过一次比较隐蔽的情况请求返回200但内容是空的。查了半天才发现是目标站点做了反爬检测到频繁请求后返回了一个空页面。解决办法是降低请求频率并且在请求头里加上Referer字段模拟从站内跳转过来的正常访问。还有一个常见问题是编码。有些站点的响应不是UTF-8直接按UTF-8解析会乱码。处理方式是在请求之后先检测编码用resp.apparent_encoding来自动判断然后再设置resp.encoding。4.2 推送内容格式错乱的处理推送格式错乱通常是因为Markdown语法和渠道的解析规则冲突。比如有些渠道不支持嵌套列表你写了嵌套结构就会显示异常。解决办法是尽量用最简单的Markdown语法只用标题、加粗、链接和普通列表不要用表格和代码块。另一个问题是特殊字符转义。产品名称里如果有方括号或者竖线会破坏Markdown的链接语法。我加了一个转义函数把可能冲突的字符替换成全角或者加上反斜杠。def escape_markdown(text): for char in [[, ], |, *, _]: text text.replace(char, \\ char) return text4.3 去重逻辑的误杀与漏杀去重逻辑调参是个细致活。误杀是指把不同的东西判成重复漏杀是指重复的东西没被去掉。误杀的危害更大因为你会漏掉重要信息。我一开始把阈值设成0.7结果“AI写作助手”和“AI绘画助手”被判成重复直接漏掉了一个品类。后来调到0.85才正常。漏杀的情况也有比如同一个产品在不同源上的标题一个是中文一个是英文编辑距离算出来相似度很低就不会被判重。这种只能靠人工规则补充比如维护一个产品名称的别名表在去重之前先做一轮别名归一化。4.4 定时任务不执行的几种可能定时任务不执行先检查进程还在不在。如果是用schedule库跑的脚本进程挂了任务自然就不跑了。建议用systemd或者supervisor来管理进程挂了能自动拉起来。如果进程在但任务没跑检查时间设置。schedule库用的是本地时间但如果你在容器里跑容器的时区可能是UTC。还有一个可能是任务执行时间太长超过了间隔时间导致下一次触发被跳过。我的处理方式是加一个锁机制任务没跑完之前不允许再次触发。问题现象可能原因排查方法解决措施采集返回空反爬拦截检查请求头和频率降低频率加Referer推送格式乱Markdown冲突检查特殊字符转义或简化语法去重误杀阈值过低抽查判重结果调高阈值至0.85任务不执行进程挂掉或时区错误检查进程和时区用进程管理工具显式设时区4.5 几个让我少走弯路的实操心得第一个心得是先跑通再优化。我一开始想把所有功能都做完美再上线结果拖了两周还没跑起来。后来改变策略先用最粗糙的方式跑通全流程哪怕采集只有一个源、推送只是纯文本先让它跑起来然后再逐步替换和优化。这样每天都有反馈调整方向也更明确。第二个心得是日志比调试器好用。这种定时任务出问题的时候往往是半夜或者清晨你不可能守着调试。把关键步骤的输入输出都记到日志里出了问题翻日志比重新跑一遍快得多。日志级别我一般设成INFO关键错误用ERROR方便过滤。第三个心得是推送内容宁少勿多。我一开始想把所有采集到的都推过去结果每天三四十条根本看不过来。后来改成只推排序后的前十五条反而每条都会认真看。信息过载比信息不足更可怕因为过载会导致你直接忽略全部。第四个心得是定期回顾和调整。这套流程不是设好就一劳永逸的数据源会变你的关注点也会变。我每个月会花十分钟看一下推送记录把不再关注的源去掉把新发现的源加进来。保持流程和需求对齐比追求技术上的完美更重要。5. 扩展方向这套流程还能怎么用这套东西跑顺了之后我发现它的核心逻辑其实可以迁移到很多场景。把“AI产品”换成任何你关注的垂直领域采集源换一下处理规则调一下推送渠道不变就是一个新的信息跟踪工具。我后来用同样的架构做了一个行业政策更新的跟踪把采集源换成几个官方发布页面处理逻辑简化成只做去重和排序推送频率改成每周一次效果也很好。另一个扩展方向是增加分析维度。现在的处理层只做了去重、排序、打标还可以加趋势分析。比如记录每个产品每天的出现次数画一个热度变化曲线当某个产品的热度突然上升时自动标记为“异动”。这个功能需要把历史数据存下来用简单的时序分析就能做不需要复杂的模型。还有一个方向是多用户支持。现在的流程是单用户的推送内容按我的偏好来排序。如果团队里多个人用可以加一个偏好配置每个人设置自己关注的标签和来源推送的时候按人过滤。这个改动主要在推送层采集和处理层可以复用。最后说一个我实际用下来觉得最有价值的点这套流程最大的收益不是省了那二十分钟而是建立了一个稳定的信息输入节奏。以前是随机刷、随机看现在是固定时间、固定格式、固定数量。节奏稳定了你对信息的消化效率会明显提升不会出现“刷了很多但什么都没记住”的情况。这个价值比自动化本身更大。