社群运营最大的困境是"沉寂"——没人说话、没人互动、最后成了死群。定时广播（早报+周报+活动通知）能维持存在感，活跃度统计能知道什么时候该推什么内容。这篇讲推送-统计-闭环三件事怎么搭。

一、定时广播：早报 + 周报 + 活动通知三类模板

定时广播分三类：每日早报（9 点推行业资讯+公司动态）、每周周报（周一推上周群活跃度数据+优秀成员）、活动通知（活动开始前 1 小时推活动详情）。三类模板各自独立存，变量用占位符，渲染时替换。

发送节奏要控——同一个群一天内不推超过 2 条、不同群的推送时间错开（9:00、9:03、9:06），避免集中发触发限流。模板里带互动引导（"评论区扣 1 领资料"），推完马上跟一条关键词自动回复接住。

二、活跃度统计：发言数 + 关键词提及率 + 活跃时段

活跃度不看群成员数，看三个指标：发言总数（每日/每周群内消息量）、关键词提及率（群内是否有业务相关关键词的讨论）、活跃时段 Top（几点发言最集中）。这三个指标直接从群消息事件里算，不用额外接口。

每日活跃度数据自动归档到 CRM，同时用于下一次的推送内容调整——活跃时段推早报、沉寂时段推活动通知。

三、闭环：推送 → 统计 → 调整推送内容

推送后自动跑统计，统计结果反馈到下一次推送的模板选择和时间调整——比如 A 群周三晚上最活跃，就把活动通知改到周三晚上推；B 群关键词提及率高，就在早报里多加行业资讯板块。

调整逻辑不用写死，运营在后台看统计数据手动调模板和时间就行，系统负责推送和统计，决策还给人。

推送-统计闭环对照

定时广播与活跃度统计实现

import time, datetime, collections, sched class GroupActivityBot: BROADCAST_PER_GROUP_PER_DAY = 2 STATS_CRON_HOUR = 22 # 每晚 22 点统计 def __init__(self, api, store, templates): self.api = api self.store = store self.templates = templates # {name: template_str} self.daily_count = collections.Counter() self.scheduler = sched.scheduler( time.time, time.sleep) self._bootstrap() def _bootstrap(self): # 早报 9:00 推所有活跃群 self.scheduler.enter( self._seconds_to_next(9, 0), 1, self.push_daily_news) # 周报周一 10:00 self.scheduler.enter( self._seconds_to_next_monday(10, 0), 1, self.push_weekly_report) # 每晚 22:00 活跃度统计 self.scheduler.enter( self._seconds_to_next(self.STATS_CRON_HOUR, 0), 1, self.run_daily_stats) import threading threading.Thread(target=self.scheduler.run, daemon=True).start() def _seconds_to_next(self, h, m): now = datetime.datetime.now() t = now.replace(hour=h, minute=m, second=0) if t <= now: t += datetime.timedelta(days=1) return (t - now).total_seconds() def _seconds_to_next_monday(self, h, m): ... def push_daily_news(self): active_groups = self.store.active_groups() for i, gid in enumerate(active_groups): if self.daily_count[gid] >= self.BROADCAST_PER_GROUP_PER_DAY: continue content = self.templates["daily_news"].format( date=datetime.date.today().isoformat(), industry=self.store.get_industry_news(limit=3))[:500] self.api.post("/group/send", { "group_id": gid, "content": content}).json() self.daily_count[gid] += 1 time.sleep(3) # 错开发送 # 次日零点清空计数 + 重排 self.scheduler.enter( self._seconds_to_next(9, 0), 1, self.push_daily_news) def push_weekly_report(self): groups = self.store.active_groups() for gid in groups: stats = self.store.group_stats(gid, days=7) content = self.templates["weekly"].format( group=gid, msg_count=stats["msg_count"], top_member=stats["top_member"], top_hour=stats["top_hour"])[:500] self.api.post("/group/send", { "group_id": gid, "content": content}).json() time.sleep(3) self.scheduler.enter( self._seconds_to_next_monday(10, 0), 1, self.push_weekly_report) # 事件入口：群消息进来时增量更新统计 def on_group_message(self, ev): self.store.incr_msg_count( ev["group_id"], ev["from"], ev["ts"]) # 关键词提及 for kw in self.store.biz_keywords(): if kw in ev["content"]: self.store.incr_keyword( ev["group_id"], kw, ev["ts"]) break # 活跃时段桶 hour = datetime.datetime.fromtimestamp( ev["ts"]).hour self.store.incr_hour(ev["group_id"], hour) def run_daily_stats(self): groups = self.store.active_groups() for gid in groups: stats = self.store.compute_stats(gid) self.store.save_stats(gid, stats, date=datetime.date.today()) # 重置日推送计数 self.daily_count.clear() self.scheduler.enter( self._seconds_to_next(self.STATS_CRON_HOUR, 0), 1, self.run_daily_stats)

落地建议

社群活跃度靠推送-统计-闭环三件事：定时广播分三类模板（早报/周报/活动通知），单群一天不超 2 条、不同群错开 3 分钟发；活跃度统计从群消息事件增量算三个指标（发言数、关键词提及率、活跃时段 Top），不用额外接口；闭环逻辑让运营手动调整推送内容和时间，系统只管推送和统计。推送用 sched 或 apscheduler 排定时，统计事件在群消息回调里增量更新，避免晚高峰拉全量群消息。Eyun 这类平台的群消息、群发送接口是底层支撑，定时调度和统计聚合在自建服务实现，接口字段以平台开发文档为准。