被暗恋的理想型看见有人加我微信不妨把“联系人通知”当工程问题设计先说一个很真实的场景你正在做一个不太紧急的需求手机微信里忽然弹出一条新的好友申请。只看了一眼头像你的判断模块瞬间过载——这个人可能是你一直在等的那个人。可现实往往更平淡等你想起来通过申请时对方已经撤回或者只剩一句“抱歉加错了”。那种感觉比上线时发现生产环境配置写错还难受。这个场景表面上是情感问题但站在技术角度看它要回答的核心问题是一条关键的信息事件能不能在正确的时间以正确的优先级触达正确的人很多人的处理方式是“多刷微信、盯得勤一点”这是典型的主动轮询方案。效率低还容易漏。真正高效的方案是把“好友申请”当作一条事件通过事件驱动的方式让系统替你做筛选和提醒谁是普通联系人谁是需要立刻响应的“理想联系人”由规则决定而不是由你的运气决定。这篇文章我会从一个完整的事件通知系统设计讲起帮你解决三类问题怎么把“有人加微信”这种社交动作抽象成一条可编程的业务事件。怎么用 Webhook、事件回调、过滤规则和消息推送搭建一个“重要联系人实时提醒”的最小闭环。在工程化部署时有哪些关于幂等、安全、隐私和消息频率的坑必须提前避开。这不是一篇教你破解微信客户端的文章也不会使用任何非官方接口。我们讨论的是一套通用的、可迁移到任意“外部事件触发业务动作”场景的事件通知架构。最终我会用 Python Flask SQLite 企业微信机器人 Webhook展示一个能跑通的最小实现。1. 为什么“谁加了你微信”也能成为一个技术问题如果只有你一个人在意这件事那确实不算技术问题。但把“你”替换成“业务系统”这台戏就完全不一样了。你可以把微信里的“好友申请”理解成一条用户行为事件。它由多个字段构成发起人标识、目标对象、申请时间、附加消息、来源渠道。你的服务器要做的不是把所有申请都同等对待而是根据预先设定的策略判断哪些申请需要立即通知负责人哪些申请只需要静默入库。在To B场景里这类需求太常见了销售线索进来需要第一时间推给对应销售。异常告警触发需要根据级别决定是否短信通知值班人。用户提交了关键工单需要马上同步给技术支持群。这些事情的本质和“暗恋对象加了你微信”完全没有区别一条优先级很高的事件不应该被淹没在普通事件流里。传统做法是让人肉去轮询。用户自己反复刷新界面运营同学定时查表技术同学定期跑脚本扫数据。问题是轮询的间隔决定了响应的延迟上限你永远不知道在两次轮询之间那条重要事件是什么时候到的。事件驱动方案的做法则完全不同系统不主动查而是被事件源告知“有新事情发生了”。事件源发起一个HTTP请求把事件数据推送到你的回调地址。你的服务只需要负责校验、过滤、落库、触发提醒。事件一旦发生通知马上发出响应速度从“取决于下一次轮询”变成“取决于网络延迟”。所以“谁加了你微信”成为一个技术问题的前提是你希望它的响应是可编程的、可自动化的而不是依赖某个人的手速和注意力。2. 场景拆解一条“好友申请”在系统里到底发生了什么我们先把整件事拆细看看一个“理想联系人申请加微信”的动作在系统视角下应该呈现成什么样子。2.1 从客户端动作到服务端事件想象你是微信服务端的架构师。用户A向用户B发送好友申请这个动作在你的系统里会产生一条事件记录大致结构是{ event_id: evt_20250101_001, event_type: new_friend_request, from_user_id: u_1001, from_user_name: 目标联系人, to_user_id: u_2002, message: 你好我是XXX, timestamp: 1735689600 }这里每个字段都有它的用途。event_id用于事件去重from_user_id用于识别发起人身份timestamp用于排序和审计。现实中的好友申请可能更复杂会有来源渠道、验证方式、标签信息等但核心结构不变它是一条带发起人和接收人的事件。2.2 事件处理的目标拿到这条事件之后我们的目标有三个层次判断这个发起人是不是需要高优先级响应的联系人。动作如果是立即触发通知如果不是只做存档。记录每一次判断和通知结果都要落库方便审计和回溯。其中最关键的是“判断”。判断逻辑不能靠拍脑袋要靠规则。规则可以是简单的联系人名单匹配也可以是稍微复杂一点的评分机制。2.3 用联系人标签替代“暗恋对象”概念你会注意到我在工程化的描述里尽量不用“暗恋对象”这种词而是用“重要联系人”或“理想联系人”。这是一个非常好的工程项目习惯把业务语言转换成数据标签。在你的系统里“是谁”不重要“打了什么标签”才重要。比如联系人数据结构[ { user_id: u_1001, nickname: agent_2025, level: ideal, alert_channels: [wecom_robot] }, { user_id: u_1002, nickname: normal_friend, level: normal, alert_channels: [] } ]后面写代码时我们的过滤逻辑只看level字段和具体是谁无关。这种解耦的好处是以后规则从“重要联系人”升级成“最近7天互动超过3次的人”只需要改匹配逻辑不需要动整条事件处理链路。3. 核心概念Webhook、事件订阅与回调通知很多新手一听到“Webhook”就紧张以为是什么复杂协议。实际上它非常简单。3.1 Webhook 是什么Webhook 是一种“反向 API”的调用方式。普通 API 是你的程序主动请求别人Webhook 是别人主动请求你的程序。对方把你的回调地址当成一个接口有事件发生就POST一条JSON数据过来。在本文的场景里你把一个可公网访问的地址交给事件源事件源检测到“有人发起好友申请”就向这个地址发送事件数据。你的服务收到数据后进入自己的处理流程。3.2 回调、事件订阅和消息推送的关系这三个词经常混用我列一个简单的对照表概念作用通俗解释事件订阅声明“我对哪些事件感兴趣”告诉事件源我只关心好友申请不关心其他消息回调地址接收事件的HTTP端点告诉事件源有事件就请求这个URL消息推送处理完后往外发送提醒告诉用户你等的联系人出现了流程是事件源在用户操作发生后根据订阅配置把事件POST到回调地址。回调地址收到后触发业务逻辑最后通过消息渠道把摘要推送给负责人。3.3 为什么不能直接改微信客户端有人会问直接写一个Hook到微信客户端里监听好友申请不就行了这是很多人最容易踩的坑。个人微信没有开放这种能力任何尝试通过非官方方式去拦截、修改某信客户端行为的手段都存在账号安全、数据合规和平台规则问题。作为CSDN的技术文章我不建议也不讲解这类非官方实现。更好的思路是把“好友申请事件源”替换成任何一个你能合法接入的平台。微信生态里企业微信提供了Webhook机器人能力方便把消息推到群里团队内部系统可以通过表单、来自渠道的事件推送完全合法地接入。本文用企业微信机器人做消息通道只是演示“事件进入系统后如何被处理”核心思路完全一致。3.4 同步还是异步收到Webhook回调后不要傻傻地在请求里把通知发送做完因为外部渠道响应慢会拖垮回调接口。推荐的做法是立刻返回200把事件写入本地队列或数据库在另一个线程里异步处理通知。这样事件源的请求快速结束你的服务不会因为某个推送渠道抖动而报错。如果你希望更规范可以引入Redis队列、消息中间件但为了最小演示一个后台线程就足够。4. 环境准备与前置条件在写代码之前先把环境理清楚。本文的示例尽量轻量依赖非常少。4.1 运行环境操作系统Windows / macOS / Linux 均可。Python 版本3.9 及以上。包管理工具pip。数据库SQLitePython 自带无需额外安装。4.2 依赖清单新建项目目录创建requirements.txtFlask2.2 requests2.28版本不要求最新以你本机实际安装为准。Flask 用来提供回调接口requests 用来调用企业微信机器人 Webhook。4.3 消息通道准备为了演示提醒效果我建议先准备一个可以向外发消息的通道。最省事的办法是创建一个企业微信群添加一个自定义机器人。在机器人配置里拿到 Webhook 地址形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key_here把your_key_here保存到环境变量或写入本地配置文件注意不要提交到 Git。如果你不想注册企业微信也可以换成任意支持标准Webhook的服务或者直接用 SMTP 发邮件。核心代码只有send_alert一个函数需要替换。4.4 项目目录结构crush-notifier/ ├── app.py ├── config.py ├── requirements.txt ├── event.db ├── data/ │ └── contacts.json └── logs/ └── notify.logconfig.py负责读取配置data/contacts.json存储联系人标签event.db是 SQLite 数据库logs/notify.log记录通知日志。5. 核心流程拆解整个“重要联系人提醒服务”可以拆成五个步骤。每一步都很小但都有魔鬼细节。5.1 步骤一定义事件回调接口你的服务需要一个HTTP端点用来接收事件源POST过来的数据。关键点有三个端点路径要固定比如/webhook/friend_request。接口必须带鉴权防止任何人伪造事件。最简单的方式是在Header中放Token。收到请求后立刻返回200把后续处理放到异步逻辑中。如果省略鉴权任何人都可以往你的回调地址POST数据然后你的企业微信就会收到一堆垃圾提醒甚至在公网被扫描工具打到崩溃。5.2 步骤二解析联系人名单和优先级当事件到达后第一件事不是急着通知而是先从事件中提取from_user_id去联系人名单里查等级。联系人名单放在data/contacts.json里结构保持简单。等级目前只区分ideal和normal。以后想加规则只需扩展这个文件配套增加匹配逻辑。5.3 步骤三过滤和匹配规则匹配逻辑是整条链路的核心。简单版本如果from_user_id存在于联系人名单且level ideal进入“高优先级通知”分支。否则只落库不通知。如果你想升级成“最近一段时间内互动次数达到阈值”的规则可以把互动行为也抽象成事件汇总后写入分析表。但这一步不建议在第一个版本里做先跑通最小闭环再逐步增加复杂度。5.4 步骤四发送提醒在高优先级分支里组装一条消息文案调用消息通道向外推送。文案要简洁但信息完整【重要联系人提醒】 你关注的用户IDu_1001 刚刚申请添加微信。 附加留言你好我是XXX。 事件时间2025-01-01 12:00:00推送动作要包在异常处理里。如果推送失败不能影响事件落库更不能让回调接口返回500。5.5 步骤五落库与审计无论是否触发通知每条事件都要写入SQLite。落库字段包括event_id事件唯一标识from_user_id发起人from_user_name发起人昵称脱敏后存储level优先级等级is_notified是否已通知create_time接收时间这一步的意义在于出了问题能查以后想做统计“每天有多少高优先级联系人请求”直接查表就行。6. 完整示例代码实现下面进入实战。我们会创建一个完整的项目代码尽量精简但结构完整。6.1 配置代码config.py创建config.pyimport os class Config: # 回调鉴权Token可由客户端写入 WEBHOOK_TOKEN os.getenv(WEBHOOK_TOKEN, change_me_to_a_random_token) # 企业微信机器人Webhook地址请替换为你的真实地址 WECOM_ROBOT_URL os.getenv( WECOM_ROBOT_URL, https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key_here, ) # SQLite数据库文件路径 DB_FILE os.getenv(DB_FILE, event.db) # 日志文件路径 LOG_FILE os.getenv(LOG_FILE, logs/notify.log)这里重点说明WEBHOOK_TOKEN不要写死在代码里用环境变量注入。企业微信机器人地址带有密钥也不能进版本库。6.2 联系人数据data/contacts.json创建data/contacts.json[ { user_id: u_1001, nickname: target_user, level: ideal, alert: true }, { user_id: u_1002, nickname: generic_friend, level: normal, alert: false } ]这只是一个演示文件。实际项目中联系人数据通常来自业务数据库或CRM系统不会用JSON硬编码。6.3 回调服务主程序app.py创建app.pyimport json import logging import sqlite3 import threading from datetime import datetime import requests from flask import Flask, jsonify, request from config import Config app Flask(__name__) # ---------- 日志初始化 ---------- def setup_logger(): import os os.makedirs(logs, exist_okTrue) logger logging.getLogger(crush-notifier) logger.setLevel(logging.INFO) fh logging.FileHandler(Config.LOG_FILE, encodingutf-8) fh.setFormatter(logging.Formatter(%(asctime)s %(levelname)s %(message)s)) logger.addHandler(fh) return logger logger setup_logger() # ---------- 数据库初始化 ---------- def init_db(): conn sqlite3.connect(Config.DB_FILE) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS event_log ( event_id TEXT PRIMARY KEY, from_user_id TEXT, from_user_name TEXT, level TEXT, is_notified INTEGER DEFAULT 0, create_time TEXT ) ) conn.commit() conn.close() # ---------- 工具函数 ---------- def load_contacts(): with open(data/contacts.json, r, encodingutf-8) as f: return json.load(f) def find_contact(from_user_id): for contact in load_contacts(): if contact[user_id] from_user_id: return contact return None def save_event_log(event_id, from_user_id, from_user_name, level, is_notified): conn sqlite3.connect(Config.DB_FILE) cursor conn.cursor() cursor.execute( INSERT OR IGNORE INTO event_log (event_id, from_user_id, from_user_name, level, is_notified, create_time) VALUES (?, ?, ?, ?, ?, ?) , ( event_id, from_user_id, from_user_name, level, 1 if is_notified else 0, datetime.now().strftime(%Y-%m-%d %H:%M:%S), ), ) conn.commit() conn.close() # ---------- 消息推送 ---------- def send_alert(contact, event): text ( 【重要联系人提醒】\n f用户ID{event.get(from_user_id)}\n f用户昵称{contact.get(nickname)}\n f附加留言{event.get(message, 无)}\n f事件时间{datetime.fromtimestamp(event.get(timestamp, 0)).isoformat()} ) payload {msgtype: text, text: {content: text}} try: resp requests.post(Config.WECOM_ROBOT_URL, jsonpayload, timeout5) resp.raise_for_status() logger.info(alert sent, event_id%s, event.get(event_id)) return True except Exception as e: logger.error(alert send failed, event_id%s, error%s, event.get(event_id), e) return False # ---------- 异步处理 ---------- def process_event(event): event_id event.get(event_id) from_user_id event.get(from_user_id) from_user_name event.get(from_user_name, unknown) contact find_contact(from_user_id) if contact and contact.get(level) ideal and contact.get(alert): notified send_alert(contact, event) save_event_log( event_id, from_user_id, from_user_name, ideal, notified ) else: save_event_log( event_id, from_user_id, from_user_name, normal, False ) logger.info(no alert needed, event_id%s, event_id) # ---------- Webhook 入口 ---------- app.route(/webhook/friend_request, methods[POST]) def webhook_friend_request(): token request.headers.get(X-Webhook-Token) if token ! Config.WEBHOOK_TOKEN: return jsonify({code: UNAUTHORIZED, message: invalid token}), 401 event request.get_json(forceTrue) if not event or not event.get(event_id): return jsonify({code: INVALID_PAYLOAD, message: event_id required}), 400 # 立即返回异步处理 thread threading.Thread(targetprocess_event, args(event,)) thread.start() return jsonify({code: OK, message: event accepted}), 200 if __name__ __main__: init_db() app.run(host0.0.0.0, port5000, debugFalse)代码需要关注几个逻辑点INSERT OR IGNORE保证了同一个event_id不会重复入库这是幂等性的基础。threading.Thread让Webhook接口立即返回不会因为推送变慢而超时。日志同时记录了收到事件和推送成功/失败方便排错。联系人匹配失败时事件仍然会落库只是is_notified为0。6.4 模拟调用curl 示例服务启动后用下面的命令模拟一条来自“ideal”联系人的好友申请curl -X POST http://127.0.0.1:5000/webhook/friend_request \ -H Content-Type: application/json \ -H X-Webhook-Token: change_me_to_a_random_token \ -d { event_id: evt_20250101_001, event_type: new_friend_request, from_user_id: u_1001, from_user_name: target_user, message: 你好我是你说的那位朋友, timestamp: 1735689600 }再模拟一条普通联系人事件curl -X POST http://127.0.0.1:5000/webhook/friend_request \ -H Content-Type: application/json \ -H X-Webhook-Token: change_me_to_a_random_token \ -d { event_id: evt_20250101_002, event_type: new_friend_request, from_user_id: u_1003, from_user_name: stranger, message: 在吗, timestamp: 1735689610 }如果你配置了企业微信机器人第一条命令执行后群里应该立刻收到“重要联系人提醒”。第二条命令不会触发任何通知。7. 运行结果与效果验证跑通这个示例不需要花很多时间但验证过程一定要完整。7.1 启动服务在项目根目录执行pip install -r requirements.txt python app.py如果看到类似下面的输出说明服务已经启动* Running on all addresses (0.0.0.0) * Running on http://127.0.0.1:50007.2 发送模拟事件并观察预期结果按照 6.4 节的 curl 命令依次发送两条事件。预期结果如下第一条事件企业微信机器人收到提醒控制台日志记录alert sent。第二条事件无提醒日志记录no alert needed。检查SQLite数据库确认有两条记录其中第一条is_notified1第二条is_notified0。查看数据库的命令sqlite3 event.db select event_id, from_user_id, level, is_notified, create_time from event_log;正常情况下的输出类似evt_20250101_001|u_1001|ideal|1|2025-01-01 12:00:01 evt_20250101_002|u_1003|normal|0|2025-01-01 12:00:027.3 如何判断成功判断这个系统是否合格不只看“接口返回200”还要看三个关键指标事件不丢失无论联系人是否匹配事件都进入日志表。重复不重复同样的event_id再次推送数据库里只有一条记录。通知不阻塞Webhook 接口响应时间应该小于100ms真正的耗时发生在异步线程里。7.4 如果失败第一步看哪里很多人跑这个示例时出问题原因集中在三处启动服务后curl 返回404或405检查 Flask 路由路径是否和 curl 一致。返回401检查X-Webhook-Token和Config.WEBHOOK_TOKEN是否一致。企业微信没收到消息检查 Webhook 地址是否完整以及企业微信机器人是否被频繁调用触发限流。8. 常见问题与排查方法问题现象可能原因排查方式解决方案curl返回401Header中的Token和代码配置不一致对比请求头与config.py中的WEBHOOK_TOKEN统一Token建议通过环境变量注入curl返回400缺少event_id或JSON格式错误检查请求体是否为合法JSON是否带event_id补齐字段重新发送请求接口返回200但数据库无记录SQLite路径和环境变量不一致检查DB_FILE配置和当前工作目录关闭服务后删除旧db文件重新init_db企业微信没收到消息Webhook地址错误、群机器人被移除、推送限流看logs/notify.log日志是否出现send failed检查地址或更换邮件等其他消息通道事件重复入库没有使用event_id做唯一约束查看event_log表主键使用INSERT OR IGNORE或先查重再插入频繁收到垃圾提醒Webhook接口没有鉴权或地址暴露公网检查日志中IP来源增加Token校验并只在测试环境使用明文HTTP这些问题的排查顺序建议固定为先看日志再看数据库最后看消息通道。logs/notify.log会把事件接收和推送结果都记录下来大多数时候问题从日志里就能定位。9. 最佳实践与工程建议最小示例能让你理解事件通知的骨架但真要部署到生产环境还有几个问题必须在设计阶段就考虑清楚。9.1 优先级规则不要写死在代码里示例代码中联系人等级写在 JSON 文件里匹配逻辑写在 Flask 服务里。短期够用长期不推荐。更好的方案是把规则抽成配置例如数据库表contact_rule字段说明rule_id规则IDmatch_field匹配字段如from_user_idmatch_value匹配值如u_1001alert_level通知级别channel消息通道enabled是否启用这样运营同学可以在后台修改规则不用动代码。9.2 回调接口必须幂等事件源可能会因为网络超时重发同一条事件。如果服务没有幂等处理用户会收到重复提醒数据库里也会出现脏数据。实现幂等的关键是event_id。每条事件必须有全局唯一ID服务收到事件时先查一次数据库如果已经存在就直接忽略。9.3 消息推送要做频率限制和降级企业微信机器人有频率限制如果短时间内触发大量高优先级事件推送会被接口拒绝。建议在推送层加一个简易的滑动窗口计数比如“一分钟内最多推送10条”。如果超限把事件标记为“待补发”而不是直接丢弃。另外消息通道要支持降级。企业微信不可用时退到邮件邮件也不可用时至少保证事件被记录到日志和数据库不让数据丢失。9.4 隐私和数据安全这类系统处理的是“谁申请加好友”等敏感信息安全底线必须清晰回调地址必须使用HTTPS防止事件内容在传输过程中被截获。Webhook Token 要用足够长的随机字符串并且定期轮换。日志中不要出现完整手机号、微信号等敏感字段一律脱敏。联系人名单和事件库务必做好访问控制不能放到公开目录。9.5 生产环境不要用 SQLiteSQLite 适合本地演示和单机小场景一旦事件量上来并发写入会成为瓶颈。生产环境建议替换为 MySQL 或 PostgreSQL同时把事件接收和事件处理拆成两个服务中间用消息队列解耦。Flask 只负责接收和返回真正的业务逻辑由消费端完成。9.6 只使用合法接口尊重平台规则这一点必须反复强调不要尝试通过非官方方式获取某信客户端的好友申请事件。个人微信不开放相关接口任何绕过客户端的行为都存在账号违规和隐私风险。企业微信机器人、服务号模板消息、第三方CRM系统的事件推送都是可以合法接入的通道。设计系统时建议优先选择平台明确开放的API。10. 总结与后续学习方向这篇博客从一个略显离谱的生活话题讲起但核心内容其实是工程上非常通用的“事件通知链路”事件源 — 回调接收 — 鉴权校验 — 规则过滤 — 异步处理 — 消息推送 — 数据落库。你只要跑通这个最小示例就会自然理解 Webhook 和轮询的区别懂得到底为什么要异步处理回调也清楚生产环境里必须考虑幂等和限流。这些经验不仅能用来做“重要联系人提醒”做告警平台、工单通知、销售线索分发时思路完全一样。如果你想继续深入可以按下面方向往下走把 SQLite 替换成 MySQL写一个更完整的表结构。用 Redis 做事件去重替代数据库主键去重。把联系人规则从 JSON 文件迁移到配置中心支持动态热更新。增加消息通道适配器同时支持企业微信、钉钉、邮件和服务号。最后提醒一句技术可以保证重要事件不再迟到但真正值得响应的人不会因为一条通知晚了几秒就消失。趁代码跑通趁消息通道正常早点把那句“你好”发出去比什么都重要。 Webhook事件通知系统实战：从好友申请提醒到工程化设计
被暗恋的理想型看见有人加我微信不妨把“联系人通知”当工程问题设计先说一个很真实的场景你正在做一个不太紧急的需求手机微信里忽然弹出一条新的好友申请。只看了一眼头像你的判断模块瞬间过载——这个人可能是你一直在等的那个人。可现实往往更平淡等你想起来通过申请时对方已经撤回或者只剩一句“抱歉加错了”。那种感觉比上线时发现生产环境配置写错还难受。这个场景表面上是情感问题但站在技术角度看它要回答的核心问题是一条关键的信息事件能不能在正确的时间以正确的优先级触达正确的人很多人的处理方式是“多刷微信、盯得勤一点”这是典型的主动轮询方案。效率低还容易漏。真正高效的方案是把“好友申请”当作一条事件通过事件驱动的方式让系统替你做筛选和提醒谁是普通联系人谁是需要立刻响应的“理想联系人”由规则决定而不是由你的运气决定。这篇文章我会从一个完整的事件通知系统设计讲起帮你解决三类问题怎么把“有人加微信”这种社交动作抽象成一条可编程的业务事件。怎么用 Webhook、事件回调、过滤规则和消息推送搭建一个“重要联系人实时提醒”的最小闭环。在工程化部署时有哪些关于幂等、安全、隐私和消息频率的坑必须提前避开。这不是一篇教你破解微信客户端的文章也不会使用任何非官方接口。我们讨论的是一套通用的、可迁移到任意“外部事件触发业务动作”场景的事件通知架构。最终我会用 Python Flask SQLite 企业微信机器人 Webhook展示一个能跑通的最小实现。1. 为什么“谁加了你微信”也能成为一个技术问题如果只有你一个人在意这件事那确实不算技术问题。但把“你”替换成“业务系统”这台戏就完全不一样了。你可以把微信里的“好友申请”理解成一条用户行为事件。它由多个字段构成发起人标识、目标对象、申请时间、附加消息、来源渠道。你的服务器要做的不是把所有申请都同等对待而是根据预先设定的策略判断哪些申请需要立即通知负责人哪些申请只需要静默入库。在To B场景里这类需求太常见了销售线索进来需要第一时间推给对应销售。异常告警触发需要根据级别决定是否短信通知值班人。用户提交了关键工单需要马上同步给技术支持群。这些事情的本质和“暗恋对象加了你微信”完全没有区别一条优先级很高的事件不应该被淹没在普通事件流里。传统做法是让人肉去轮询。用户自己反复刷新界面运营同学定时查表技术同学定期跑脚本扫数据。问题是轮询的间隔决定了响应的延迟上限你永远不知道在两次轮询之间那条重要事件是什么时候到的。事件驱动方案的做法则完全不同系统不主动查而是被事件源告知“有新事情发生了”。事件源发起一个HTTP请求把事件数据推送到你的回调地址。你的服务只需要负责校验、过滤、落库、触发提醒。事件一旦发生通知马上发出响应速度从“取决于下一次轮询”变成“取决于网络延迟”。所以“谁加了你微信”成为一个技术问题的前提是你希望它的响应是可编程的、可自动化的而不是依赖某个人的手速和注意力。2. 场景拆解一条“好友申请”在系统里到底发生了什么我们先把整件事拆细看看一个“理想联系人申请加微信”的动作在系统视角下应该呈现成什么样子。2.1 从客户端动作到服务端事件想象你是微信服务端的架构师。用户A向用户B发送好友申请这个动作在你的系统里会产生一条事件记录大致结构是{ event_id: evt_20250101_001, event_type: new_friend_request, from_user_id: u_1001, from_user_name: 目标联系人, to_user_id: u_2002, message: 你好我是XXX, timestamp: 1735689600 }这里每个字段都有它的用途。event_id用于事件去重from_user_id用于识别发起人身份timestamp用于排序和审计。现实中的好友申请可能更复杂会有来源渠道、验证方式、标签信息等但核心结构不变它是一条带发起人和接收人的事件。2.2 事件处理的目标拿到这条事件之后我们的目标有三个层次判断这个发起人是不是需要高优先级响应的联系人。动作如果是立即触发通知如果不是只做存档。记录每一次判断和通知结果都要落库方便审计和回溯。其中最关键的是“判断”。判断逻辑不能靠拍脑袋要靠规则。规则可以是简单的联系人名单匹配也可以是稍微复杂一点的评分机制。2.3 用联系人标签替代“暗恋对象”概念你会注意到我在工程化的描述里尽量不用“暗恋对象”这种词而是用“重要联系人”或“理想联系人”。这是一个非常好的工程项目习惯把业务语言转换成数据标签。在你的系统里“是谁”不重要“打了什么标签”才重要。比如联系人数据结构[ { user_id: u_1001, nickname: agent_2025, level: ideal, alert_channels: [wecom_robot] }, { user_id: u_1002, nickname: normal_friend, level: normal, alert_channels: [] } ]后面写代码时我们的过滤逻辑只看level字段和具体是谁无关。这种解耦的好处是以后规则从“重要联系人”升级成“最近7天互动超过3次的人”只需要改匹配逻辑不需要动整条事件处理链路。3. 核心概念Webhook、事件订阅与回调通知很多新手一听到“Webhook”就紧张以为是什么复杂协议。实际上它非常简单。3.1 Webhook 是什么Webhook 是一种“反向 API”的调用方式。普通 API 是你的程序主动请求别人Webhook 是别人主动请求你的程序。对方把你的回调地址当成一个接口有事件发生就POST一条JSON数据过来。在本文的场景里你把一个可公网访问的地址交给事件源事件源检测到“有人发起好友申请”就向这个地址发送事件数据。你的服务收到数据后进入自己的处理流程。3.2 回调、事件订阅和消息推送的关系这三个词经常混用我列一个简单的对照表概念作用通俗解释事件订阅声明“我对哪些事件感兴趣”告诉事件源我只关心好友申请不关心其他消息回调地址接收事件的HTTP端点告诉事件源有事件就请求这个URL消息推送处理完后往外发送提醒告诉用户你等的联系人出现了流程是事件源在用户操作发生后根据订阅配置把事件POST到回调地址。回调地址收到后触发业务逻辑最后通过消息渠道把摘要推送给负责人。3.3 为什么不能直接改微信客户端有人会问直接写一个Hook到微信客户端里监听好友申请不就行了这是很多人最容易踩的坑。个人微信没有开放这种能力任何尝试通过非官方方式去拦截、修改某信客户端行为的手段都存在账号安全、数据合规和平台规则问题。作为CSDN的技术文章我不建议也不讲解这类非官方实现。更好的思路是把“好友申请事件源”替换成任何一个你能合法接入的平台。微信生态里企业微信提供了Webhook机器人能力方便把消息推到群里团队内部系统可以通过表单、来自渠道的事件推送完全合法地接入。本文用企业微信机器人做消息通道只是演示“事件进入系统后如何被处理”核心思路完全一致。3.4 同步还是异步收到Webhook回调后不要傻傻地在请求里把通知发送做完因为外部渠道响应慢会拖垮回调接口。推荐的做法是立刻返回200把事件写入本地队列或数据库在另一个线程里异步处理通知。这样事件源的请求快速结束你的服务不会因为某个推送渠道抖动而报错。如果你希望更规范可以引入Redis队列、消息中间件但为了最小演示一个后台线程就足够。4. 环境准备与前置条件在写代码之前先把环境理清楚。本文的示例尽量轻量依赖非常少。4.1 运行环境操作系统Windows / macOS / Linux 均可。Python 版本3.9 及以上。包管理工具pip。数据库SQLitePython 自带无需额外安装。4.2 依赖清单新建项目目录创建requirements.txtFlask2.2 requests2.28版本不要求最新以你本机实际安装为准。Flask 用来提供回调接口requests 用来调用企业微信机器人 Webhook。4.3 消息通道准备为了演示提醒效果我建议先准备一个可以向外发消息的通道。最省事的办法是创建一个企业微信群添加一个自定义机器人。在机器人配置里拿到 Webhook 地址形如https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key_here把your_key_here保存到环境变量或写入本地配置文件注意不要提交到 Git。如果你不想注册企业微信也可以换成任意支持标准Webhook的服务或者直接用 SMTP 发邮件。核心代码只有send_alert一个函数需要替换。4.4 项目目录结构crush-notifier/ ├── app.py ├── config.py ├── requirements.txt ├── event.db ├── data/ │ └── contacts.json └── logs/ └── notify.logconfig.py负责读取配置data/contacts.json存储联系人标签event.db是 SQLite 数据库logs/notify.log记录通知日志。5. 核心流程拆解整个“重要联系人提醒服务”可以拆成五个步骤。每一步都很小但都有魔鬼细节。5.1 步骤一定义事件回调接口你的服务需要一个HTTP端点用来接收事件源POST过来的数据。关键点有三个端点路径要固定比如/webhook/friend_request。接口必须带鉴权防止任何人伪造事件。最简单的方式是在Header中放Token。收到请求后立刻返回200把后续处理放到异步逻辑中。如果省略鉴权任何人都可以往你的回调地址POST数据然后你的企业微信就会收到一堆垃圾提醒甚至在公网被扫描工具打到崩溃。5.2 步骤二解析联系人名单和优先级当事件到达后第一件事不是急着通知而是先从事件中提取from_user_id去联系人名单里查等级。联系人名单放在data/contacts.json里结构保持简单。等级目前只区分ideal和normal。以后想加规则只需扩展这个文件配套增加匹配逻辑。5.3 步骤三过滤和匹配规则匹配逻辑是整条链路的核心。简单版本如果from_user_id存在于联系人名单且level ideal进入“高优先级通知”分支。否则只落库不通知。如果你想升级成“最近一段时间内互动次数达到阈值”的规则可以把互动行为也抽象成事件汇总后写入分析表。但这一步不建议在第一个版本里做先跑通最小闭环再逐步增加复杂度。5.4 步骤四发送提醒在高优先级分支里组装一条消息文案调用消息通道向外推送。文案要简洁但信息完整【重要联系人提醒】 你关注的用户IDu_1001 刚刚申请添加微信。 附加留言你好我是XXX。 事件时间2025-01-01 12:00:00推送动作要包在异常处理里。如果推送失败不能影响事件落库更不能让回调接口返回500。5.5 步骤五落库与审计无论是否触发通知每条事件都要写入SQLite。落库字段包括event_id事件唯一标识from_user_id发起人from_user_name发起人昵称脱敏后存储level优先级等级is_notified是否已通知create_time接收时间这一步的意义在于出了问题能查以后想做统计“每天有多少高优先级联系人请求”直接查表就行。6. 完整示例代码实现下面进入实战。我们会创建一个完整的项目代码尽量精简但结构完整。6.1 配置代码config.py创建config.pyimport os class Config: # 回调鉴权Token可由客户端写入 WEBHOOK_TOKEN os.getenv(WEBHOOK_TOKEN, change_me_to_a_random_token) # 企业微信机器人Webhook地址请替换为你的真实地址 WECOM_ROBOT_URL os.getenv( WECOM_ROBOT_URL, https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key_here, ) # SQLite数据库文件路径 DB_FILE os.getenv(DB_FILE, event.db) # 日志文件路径 LOG_FILE os.getenv(LOG_FILE, logs/notify.log)这里重点说明WEBHOOK_TOKEN不要写死在代码里用环境变量注入。企业微信机器人地址带有密钥也不能进版本库。6.2 联系人数据data/contacts.json创建data/contacts.json[ { user_id: u_1001, nickname: target_user, level: ideal, alert: true }, { user_id: u_1002, nickname: generic_friend, level: normal, alert: false } ]这只是一个演示文件。实际项目中联系人数据通常来自业务数据库或CRM系统不会用JSON硬编码。6.3 回调服务主程序app.py创建app.pyimport json import logging import sqlite3 import threading from datetime import datetime import requests from flask import Flask, jsonify, request from config import Config app Flask(__name__) # ---------- 日志初始化 ---------- def setup_logger(): import os os.makedirs(logs, exist_okTrue) logger logging.getLogger(crush-notifier) logger.setLevel(logging.INFO) fh logging.FileHandler(Config.LOG_FILE, encodingutf-8) fh.setFormatter(logging.Formatter(%(asctime)s %(levelname)s %(message)s)) logger.addHandler(fh) return logger logger setup_logger() # ---------- 数据库初始化 ---------- def init_db(): conn sqlite3.connect(Config.DB_FILE) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS event_log ( event_id TEXT PRIMARY KEY, from_user_id TEXT, from_user_name TEXT, level TEXT, is_notified INTEGER DEFAULT 0, create_time TEXT ) ) conn.commit() conn.close() # ---------- 工具函数 ---------- def load_contacts(): with open(data/contacts.json, r, encodingutf-8) as f: return json.load(f) def find_contact(from_user_id): for contact in load_contacts(): if contact[user_id] from_user_id: return contact return None def save_event_log(event_id, from_user_id, from_user_name, level, is_notified): conn sqlite3.connect(Config.DB_FILE) cursor conn.cursor() cursor.execute( INSERT OR IGNORE INTO event_log (event_id, from_user_id, from_user_name, level, is_notified, create_time) VALUES (?, ?, ?, ?, ?, ?) , ( event_id, from_user_id, from_user_name, level, 1 if is_notified else 0, datetime.now().strftime(%Y-%m-%d %H:%M:%S), ), ) conn.commit() conn.close() # ---------- 消息推送 ---------- def send_alert(contact, event): text ( 【重要联系人提醒】\n f用户ID{event.get(from_user_id)}\n f用户昵称{contact.get(nickname)}\n f附加留言{event.get(message, 无)}\n f事件时间{datetime.fromtimestamp(event.get(timestamp, 0)).isoformat()} ) payload {msgtype: text, text: {content: text}} try: resp requests.post(Config.WECOM_ROBOT_URL, jsonpayload, timeout5) resp.raise_for_status() logger.info(alert sent, event_id%s, event.get(event_id)) return True except Exception as e: logger.error(alert send failed, event_id%s, error%s, event.get(event_id), e) return False # ---------- 异步处理 ---------- def process_event(event): event_id event.get(event_id) from_user_id event.get(from_user_id) from_user_name event.get(from_user_name, unknown) contact find_contact(from_user_id) if contact and contact.get(level) ideal and contact.get(alert): notified send_alert(contact, event) save_event_log( event_id, from_user_id, from_user_name, ideal, notified ) else: save_event_log( event_id, from_user_id, from_user_name, normal, False ) logger.info(no alert needed, event_id%s, event_id) # ---------- Webhook 入口 ---------- app.route(/webhook/friend_request, methods[POST]) def webhook_friend_request(): token request.headers.get(X-Webhook-Token) if token ! Config.WEBHOOK_TOKEN: return jsonify({code: UNAUTHORIZED, message: invalid token}), 401 event request.get_json(forceTrue) if not event or not event.get(event_id): return jsonify({code: INVALID_PAYLOAD, message: event_id required}), 400 # 立即返回异步处理 thread threading.Thread(targetprocess_event, args(event,)) thread.start() return jsonify({code: OK, message: event accepted}), 200 if __name__ __main__: init_db() app.run(host0.0.0.0, port5000, debugFalse)代码需要关注几个逻辑点INSERT OR IGNORE保证了同一个event_id不会重复入库这是幂等性的基础。threading.Thread让Webhook接口立即返回不会因为推送变慢而超时。日志同时记录了收到事件和推送成功/失败方便排错。联系人匹配失败时事件仍然会落库只是is_notified为0。6.4 模拟调用curl 示例服务启动后用下面的命令模拟一条来自“ideal”联系人的好友申请curl -X POST http://127.0.0.1:5000/webhook/friend_request \ -H Content-Type: application/json \ -H X-Webhook-Token: change_me_to_a_random_token \ -d { event_id: evt_20250101_001, event_type: new_friend_request, from_user_id: u_1001, from_user_name: target_user, message: 你好我是你说的那位朋友, timestamp: 1735689600 }再模拟一条普通联系人事件curl -X POST http://127.0.0.1:5000/webhook/friend_request \ -H Content-Type: application/json \ -H X-Webhook-Token: change_me_to_a_random_token \ -d { event_id: evt_20250101_002, event_type: new_friend_request, from_user_id: u_1003, from_user_name: stranger, message: 在吗, timestamp: 1735689610 }如果你配置了企业微信机器人第一条命令执行后群里应该立刻收到“重要联系人提醒”。第二条命令不会触发任何通知。7. 运行结果与效果验证跑通这个示例不需要花很多时间但验证过程一定要完整。7.1 启动服务在项目根目录执行pip install -r requirements.txt python app.py如果看到类似下面的输出说明服务已经启动* Running on all addresses (0.0.0.0) * Running on http://127.0.0.1:50007.2 发送模拟事件并观察预期结果按照 6.4 节的 curl 命令依次发送两条事件。预期结果如下第一条事件企业微信机器人收到提醒控制台日志记录alert sent。第二条事件无提醒日志记录no alert needed。检查SQLite数据库确认有两条记录其中第一条is_notified1第二条is_notified0。查看数据库的命令sqlite3 event.db select event_id, from_user_id, level, is_notified, create_time from event_log;正常情况下的输出类似evt_20250101_001|u_1001|ideal|1|2025-01-01 12:00:01 evt_20250101_002|u_1003|normal|0|2025-01-01 12:00:027.3 如何判断成功判断这个系统是否合格不只看“接口返回200”还要看三个关键指标事件不丢失无论联系人是否匹配事件都进入日志表。重复不重复同样的event_id再次推送数据库里只有一条记录。通知不阻塞Webhook 接口响应时间应该小于100ms真正的耗时发生在异步线程里。7.4 如果失败第一步看哪里很多人跑这个示例时出问题原因集中在三处启动服务后curl 返回404或405检查 Flask 路由路径是否和 curl 一致。返回401检查X-Webhook-Token和Config.WEBHOOK_TOKEN是否一致。企业微信没收到消息检查 Webhook 地址是否完整以及企业微信机器人是否被频繁调用触发限流。8. 常见问题与排查方法问题现象可能原因排查方式解决方案curl返回401Header中的Token和代码配置不一致对比请求头与config.py中的WEBHOOK_TOKEN统一Token建议通过环境变量注入curl返回400缺少event_id或JSON格式错误检查请求体是否为合法JSON是否带event_id补齐字段重新发送请求接口返回200但数据库无记录SQLite路径和环境变量不一致检查DB_FILE配置和当前工作目录关闭服务后删除旧db文件重新init_db企业微信没收到消息Webhook地址错误、群机器人被移除、推送限流看logs/notify.log日志是否出现send failed检查地址或更换邮件等其他消息通道事件重复入库没有使用event_id做唯一约束查看event_log表主键使用INSERT OR IGNORE或先查重再插入频繁收到垃圾提醒Webhook接口没有鉴权或地址暴露公网检查日志中IP来源增加Token校验并只在测试环境使用明文HTTP这些问题的排查顺序建议固定为先看日志再看数据库最后看消息通道。logs/notify.log会把事件接收和推送结果都记录下来大多数时候问题从日志里就能定位。9. 最佳实践与工程建议最小示例能让你理解事件通知的骨架但真要部署到生产环境还有几个问题必须在设计阶段就考虑清楚。9.1 优先级规则不要写死在代码里示例代码中联系人等级写在 JSON 文件里匹配逻辑写在 Flask 服务里。短期够用长期不推荐。更好的方案是把规则抽成配置例如数据库表contact_rule字段说明rule_id规则IDmatch_field匹配字段如from_user_idmatch_value匹配值如u_1001alert_level通知级别channel消息通道enabled是否启用这样运营同学可以在后台修改规则不用动代码。9.2 回调接口必须幂等事件源可能会因为网络超时重发同一条事件。如果服务没有幂等处理用户会收到重复提醒数据库里也会出现脏数据。实现幂等的关键是event_id。每条事件必须有全局唯一ID服务收到事件时先查一次数据库如果已经存在就直接忽略。9.3 消息推送要做频率限制和降级企业微信机器人有频率限制如果短时间内触发大量高优先级事件推送会被接口拒绝。建议在推送层加一个简易的滑动窗口计数比如“一分钟内最多推送10条”。如果超限把事件标记为“待补发”而不是直接丢弃。另外消息通道要支持降级。企业微信不可用时退到邮件邮件也不可用时至少保证事件被记录到日志和数据库不让数据丢失。9.4 隐私和数据安全这类系统处理的是“谁申请加好友”等敏感信息安全底线必须清晰回调地址必须使用HTTPS防止事件内容在传输过程中被截获。Webhook Token 要用足够长的随机字符串并且定期轮换。日志中不要出现完整手机号、微信号等敏感字段一律脱敏。联系人名单和事件库务必做好访问控制不能放到公开目录。9.5 生产环境不要用 SQLiteSQLite 适合本地演示和单机小场景一旦事件量上来并发写入会成为瓶颈。生产环境建议替换为 MySQL 或 PostgreSQL同时把事件接收和事件处理拆成两个服务中间用消息队列解耦。Flask 只负责接收和返回真正的业务逻辑由消费端完成。9.6 只使用合法接口尊重平台规则这一点必须反复强调不要尝试通过非官方方式获取某信客户端的好友申请事件。个人微信不开放相关接口任何绕过客户端的行为都存在账号违规和隐私风险。企业微信机器人、服务号模板消息、第三方CRM系统的事件推送都是可以合法接入的通道。设计系统时建议优先选择平台明确开放的API。10. 总结与后续学习方向这篇博客从一个略显离谱的生活话题讲起但核心内容其实是工程上非常通用的“事件通知链路”事件源 — 回调接收 — 鉴权校验 — 规则过滤 — 异步处理 — 消息推送 — 数据落库。你只要跑通这个最小示例就会自然理解 Webhook 和轮询的区别懂得到底为什么要异步处理回调也清楚生产环境里必须考虑幂等和限流。这些经验不仅能用来做“重要联系人提醒”做告警平台、工单通知、销售线索分发时思路完全一样。如果你想继续深入可以按下面方向往下走把 SQLite 替换成 MySQL写一个更完整的表结构。用 Redis 做事件去重替代数据库主键去重。把联系人规则从 JSON 文件迁移到配置中心支持动态热更新。增加消息通道适配器同时支持企业微信、钉钉、邮件和服务号。最后提醒一句技术可以保证重要事件不再迟到但真正值得响应的人不会因为一条通知晚了几秒就消失。趁代码跑通趁消息通道正常早点把那句“你好”发出去比什么都重要。