前置群列表运营速览摘要

变更项:
1. 新增群列表层批量运营速览接口,输出健康度、核心成员、待激活成员和简短运营摘要。
2. 在通讯录管理的群组列表中增加运营速览列,把成员分层摘要信号前置到列表层。
3. 保留详情页原有完整群洞察内容,不将完整成员明细直接搬到列表层,继续采用列表筛群、详情看明细的结构。
This commit is contained in:
liuwei
2026-05-06 11:50:11 +08:00
parent b618bcc30d
commit 63d5a5e716
2 changed files with 188 additions and 1 deletions

View File

@@ -334,6 +334,84 @@ def _build_ops_action_cards(group_name: str, ops_profile: dict, ops_members: dic
return action_cards[:4]
def _build_group_list_ops_preview(server, group_id: str, group_name: str) -> dict:
"""构建群列表层可展示的轻量运营速览。
设计原则:
1. 列表页只放“筛选信号”,不放完整成员明细,避免信息密度过高;
2. 详情页里已经有完整诊断,因此这里优先输出健康度、核心成员、待激活成员等摘要;
3. 为了控制接口成本,这里不复用重型画像构建,只做轻量聚合和短摘要生成。
"""
now = datetime.now()
last_30d = now - timedelta(days=30)
member_summary = server.contact_db.get_group_member_summary(group_id, inactive_days=30)
message_stats_30d = server.message_storage.get_message_stats_by_date_range(group_id, last_30d, now)
plugin_summary = server.stats_db.get_group_plugin_summary(group_id, days=30)
speaker_ranking = server.message_storage.get_group_member_message_ranking(group_id, last_30d, now, limit=3)
activation_candidates = _load_activation_candidates(server, group_id, limit=3)
recent_summaries = _load_recent_group_summaries(server, group_id, limit=1)
total_members = member_summary.get("in_group_members") or member_summary.get("total_members") or 0
active_30d = int(member_summary.get("active_30d_members") or 0)
zombie_count = int(member_summary.get("inactive_members") or 0)
msg_30d_total = int(message_stats_30d.get("total_count") or 0)
plugin_total_calls = int(plugin_summary.get("total_calls") or 0)
# 列表页也复用和详情页一致的健康度口径:
# 1. 这样用户从列表点进详情时,数值语义不会跳变;
# 2. 逻辑保持一套,后续调分只改一处思路更稳;
# 3. 列表上只展示分值,不额外塞入复杂推导过程。
health_score = 100
if total_members > 0:
health_score -= min(35, round((zombie_count / total_members) * 35))
health_score -= min(25, round((1 - min(active_30d / total_members, 1)) * 25))
if msg_30d_total == 0:
health_score -= 25
elif msg_30d_total < 50:
health_score -= 12
if plugin_total_calls == 0:
health_score -= 10
health_score = max(0, min(100, health_score))
core_member_names = []
for item in speaker_ranking:
sender = str(item.get("sender") or "").strip()
display_name = server.contact_manager.get_group_name(group_id, sender) or sender
if display_name:
core_member_names.append(display_name)
activation_names = [
str(item.get("display_name") or "").strip()
for item in activation_candidates
if str(item.get("display_name") or "").strip()
]
summary_excerpt = ""
if recent_summaries:
summary_excerpt = str(recent_summaries[0].get("summary_excerpt") or "").strip()
if not summary_excerpt:
if core_member_names:
summary_excerpt = f"核心讨论主要由 {''.join(core_member_names[:2])} 带动。"
elif plugin_total_calls > 0:
summary_excerpt = f"近30天已有 {plugin_total_calls} 次插件调用,可继续做功能渗透。"
else:
summary_excerpt = "当前运营数据较少,建议先观察活跃与功能渗透。"
return {
"group_id": group_id,
"group_name": group_name,
"health_score": int(health_score),
"active_member_count_30d": active_30d,
"zombie_member_count": zombie_count,
"core_member_names": core_member_names[:3],
"activation_candidate_count": len(activation_names),
"activation_candidate_names": activation_names[:3],
"plugin_call_count_30d": plugin_total_calls,
"summary_excerpt": summary_excerpt[:120] + ("..." if len(summary_excerpt) > 120 else ""),
}
# 旧群权限页入口兼容跳转
@robot_bp.route('/')
@login_required
@@ -547,6 +625,41 @@ def api_group_message_trend(group_id):
return jsonify({'success': False, 'error': str(e)}), 500
@robot_bp.route('/api/groups/ops_preview', methods=['POST'])
@login_required
def api_groups_ops_preview():
"""批量获取群列表层的运营速览摘要。"""
try:
server = current_app.dashboard_server
data = request.json or {}
group_ids = data.get("group_ids") or []
# 列表批量接口只接受有限数量群组:
# 1. 避免一次把全量群的详情聚合全算一遍;
# 2. 通讯录列表本身就有分页/筛选,首版以“当前批次可见群”为主;
# 3. 这里做硬上限,防止误调用把后台拖慢。
cleaned_group_ids = []
for item in group_ids:
group_id = str(item or "").strip()
if group_id and group_id not in cleaned_group_ids:
cleaned_group_ids.append(group_id)
cleaned_group_ids = cleaned_group_ids[:30]
preview_map = {}
for group_id in cleaned_group_ids:
group_name = server.contact_manager.get_nickname(group_id) or group_id
preview_map[group_id] = _build_group_list_ops_preview(server, group_id, group_name)
return jsonify({
"success": True,
"data": {
"preview_map": preview_map,
}
})
except Exception as e:
LOG.error(f"批量获取群运营速览失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@robot_bp.route('/api/group/<group_id>/detail')
@login_required
def api_group_detail(group_id):