每天要做的事其实就三件早上批量处理申请打标签、上班时间盯 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——早上存一份、晚上存一份对比差异就知道谁加了谁删了。每日周期任务对照时段任务接口频率/节奏9:00批量通过来源标签/friend/accept、/friend/update0.3/s10-18 点FAQ 关键词回复/send实时监听18:00聊天总结异动入库/friend/list、消息事件每日 1 次每日周期调度实现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(targetself.scheduler.run, daemonTrue).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(targetself.faq_loop, daemonTrue).start() def _seconds_to_next(self, h, m): now datetime.datetime.now() target now.replace(hourh, minutem, second0) if target now: target datetime.timedelta(days1) 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 逻辑在自建服务实现接口字段以平台开发文档为准。
