每天要做的事其实就三件：早上批量处理申请+打标签、上班时间盯 FAQ 自动回复、下班后把当天的聊天和好友异动沉淀入库。把这三件事编排进一个每日调度，能省掉至少 2 小时的重复劳动。

一、早 9 点：批量通过 + 来源标签

每天早上 9 点，程序把前一天累积的好友申请全部拉出来批量通过。通过的节奏控制在每秒 0.3 个，同时按申请来源（二维码、名片推荐、搜索、群聊）自动打对应标签——二维码加的打"活动渠道"、名片推的打"转介绍"。

来源字段在好友申请事件里已经带了（source 字段取值：qr、referral、search、group），不用额外调接口查。通过后延迟 5-10 秒发对应来源的欢迎语模板，别秒回。

二、10-18 点：FAQ 关键词回复

上班时间（10:00-18:00）程序保持运行，消息进来先跑一轮 FAQ 关键词匹配（精确词库+正则+模糊三级）。命中直接答，没命中转给人工客服。匹配结果要带时间戳和发送人，方便晚上复盘时统计热门问题。

FAQ 词库放在可配置文件里，新增或修改关键词不用改代码，运营自己在配置文件里改一下就行。

三、晚 6 点：聊天总结 + 好友异动入库

下班后程序自动做两件事：把当天所有消息（收发双向）落库、把当天新增/删除/拉黑的好友异动同步到 CRM。落库时按客户 wxid 分组、按时间排序，方便后续检索。

好友异动通过两次拉取好友列表做 diff——早上存一份、晚上存一份，对比差异就知道谁加了谁删了。

每日周期任务对照

每日周期调度实现

import time, sched, threading, queue, datetime class DailyPipeline: def __init__(self, api, store, faq_config): self.api = api self.store = store self.faq = faq_config # {keyword: answer} self.scheduler = sched.scheduler( time.time, time.sleep) self._schedules() threading.Thread(target=self.scheduler.run, daemon=True).start() def _schedules(self): # 首次排程 + 循环重排 self.scheduler.enter( self._seconds_to_next(9, 0), 1, self.morning_batch) self.scheduler.enter( self._seconds_to_next(18, 0), 1, self.evening_sync) # FAQ 实时监听独立线程 threading.Thread(target=self.faq_loop, daemon=True).start() def _seconds_to_next(self, h, m): now = datetime.datetime.now() target = now.replace(hour=h, minute=m, second=0) if target <= now: target += datetime.timedelta(days=1) return (target - now).total_seconds() # 早 9 点：批量通过 def morning_batch(self): reqs = self.api.get("/friend/pending").json()["items"] for req in reqs: self.api.post("/friend/accept", {"req_id": req["req_id"]}).json() # 按来源打标签 source_tag = self._source_to_tag(req.get("source")) if source_tag: self.api.post("/friend/update", {"wxid": req["wxid"], "tag_list": [source_tag]}).json() # 延迟发欢迎语 time.sleep(8) self.api.post("/send", { "wxid": req["wxid"], "content": self.store.welcome_for( req.get("source", "general"))}).json() time.sleep(1 / 0.3) # 0.3/s 控速 # 排下一次 self.scheduler.enter( self._seconds_to_next(9, 0), 1, self.morning_batch) def _source_to_tag(self, source): return {"qr": "活动渠道", "referral": "转介绍", "search": "搜索添加", "group": "群聊加入"}.get(source) # 10-18 点 FAQ 监听 def faq_loop(self): while True: now = datetime.datetime.now().hour if 10 <= now < 18: events = self.api.get("/message/poll", params={"since": self.store.last_msg_id()}).json() for ev in events: if ev.get("from_me"): continue answer = self._match_faq(ev["content"]) if answer: self.api.post("/send", { "receiver": ev["from"], "content": answer}).json() else: self.store.push_to_human(ev) time.sleep(3) def _match_faq(self, text): for kw, ans in self.faq.items(): if kw in text: return ans return None # 晚 6 点：聊天总结 + 异动入库 def evening_sync(self): msgs = self.api.get("/message/list", params={"date": datetime.date.today()}).json()["items"] self.store.batch_save_msgs(msgs) # 好友异动 diff self.store.friend_list_new = self.api.get("/friend/list").json() diff = self.store.diff_friends() self.store.save_friend_diff(diff) self.scheduler.enter( self._seconds_to_next(18, 0), 1, self.evening_sync)

落地建议

每日三件事用 sched 或 apscheduler 排好触发时间，不用起 cron。早 9 点批量通过控 0.3 每秒、同时按来源打标签，欢迎语延迟 5-10 秒别秒回。FAQ 关键词放可配置文件，运营自己改不用重部署。晚上 6 点做两件事：当天消息全量落库、好友列表和早上做 diff 拿异动。Eyun这类平台的消息、好友接口是底层支撑，每日编排和落库 diff 逻辑在自建服务实现，接口字段以平台开发文档为准。