feat: add group detail dashboard insights

db/contacts_db.py

@@ -743,3 +743,64 @@ class ContactsDBOperator(BaseDBOperator):
         except Exception as e:
             self.LOG.error(f"获取群{chatroom_id}潜水排行失败: {e}")
             return []
+
+    def get_group_member_summary(self, chatroom_id: str, inactive_days: int = 30) -> Dict[str, Any]:
+        """获取群成员概览摘要"""
+        try:
+            sql = """
+            SELECT
+                COUNT(*) AS total_members,
+                SUM(CASE WHEN status = 1 THEN 1 ELSE 0 END) AS in_group_members,
+                SUM(CASE WHEN is_owner = 1 THEN 1 ELSE 0 END) AS owner_count,
+                SUM(CASE WHEN is_admin = 1 THEN 1 ELSE 0 END) AS admin_count,
+                SUM(CASE WHEN latest_active_time IS NOT NULL THEN 1 ELSE 0 END) AS spoken_members,
+                SUM(CASE WHEN latest_active_time IS NULL THEN 1 ELSE 0 END) AS never_spoken_members,
+                SUM(
+                    CASE
+                        WHEN latest_active_time IS NOT NULL
+                         AND latest_active_time >= DATE_SUB(NOW(), INTERVAL 7 DAY)
+                        THEN 1 ELSE 0
+                    END
+                ) AS active_7d_members,
+                SUM(
+                    CASE
+                        WHEN latest_active_time IS NOT NULL
+                         AND latest_active_time >= DATE_SUB(NOW(), INTERVAL 30 DAY)
+                        THEN 1 ELSE 0
+                    END
+                ) AS active_30d_members,
+                SUM(
+                    CASE
+                        WHEN latest_active_time IS NULL
+                          OR latest_active_time < DATE_SUB(NOW(), INTERVAL %s DAY)
+                        THEN 1 ELSE 0
+                    END
+                ) AS inactive_members
+            FROM t_chatroom_member
+            WHERE chatroom_id = %s
+            """
+            result = self.execute_query(sql, (inactive_days, chatroom_id), fetch_one=True) or {}
+            return {
+                "total_members": int(result.get("total_members") or 0),
+                "in_group_members": int(result.get("in_group_members") or 0),
+                "owner_count": int(result.get("owner_count") or 0),
+                "admin_count": int(result.get("admin_count") or 0),
+                "spoken_members": int(result.get("spoken_members") or 0),
+                "never_spoken_members": int(result.get("never_spoken_members") or 0),
+                "active_7d_members": int(result.get("active_7d_members") or 0),
+                "active_30d_members": int(result.get("active_30d_members") or 0),
+                "inactive_members": int(result.get("inactive_members") or 0),
+            }
+        except Exception as e:
+            self.LOG.error(f"获取群{chatroom_id}成员摘要失败: {e}")
+            return {
+                "total_members": 0,
+                "in_group_members": 0,
+                "owner_count": 0,
+                "admin_count": 0,
+                "spoken_members": 0,
+                "never_spoken_members": 0,
+                "active_7d_members": 0,
+                "active_30d_members": 0,
+                "inactive_members": 0,
+            }

db/message_storage.py

@@ -269,6 +269,72 @@ class MessageStorageDB(BaseDBOperator):
         """
         return self.execute_query(sql, (group_id, days)) or []
 
+    def get_group_member_message_ranking(self, group_id: str, start_time: datetime,
+                                         end_time: datetime, limit: int = 10) -> List[Dict]:
+        """获取群成员发言排行"""
+        sql = """
+        SELECT
+            sender,
+            COUNT(*) AS message_count,
+            MAX(timestamp) AS last_message_time
+        FROM messages
+        WHERE timestamp >= %s
+          AND timestamp <= %s
+          AND group_id = %s
+          AND sender IS NOT NULL
+          AND sender <> ''
+        GROUP BY sender
+        ORDER BY message_count DESC, last_message_time DESC
+        LIMIT %s
+        """
+        params = (
+            start_time.strftime('%Y-%m-%d %H:%M:%S'),
+            end_time.strftime('%Y-%m-%d %H:%M:%S'),
+            group_id,
+            limit,
+        )
+        rows = self.execute_query(sql, params) or []
+        for row in rows:
+            dt = row.get("last_message_time")
+            if isinstance(dt, datetime):
+                row["last_message_time"] = dt.strftime("%Y-%m-%d %H:%M:%S")
+        return rows
+
+    def get_group_hourly_distribution(self, group_id: str, days: int = 30) -> List[Dict]:
+        """获取群消息小时分布"""
+        sql = """
+        SELECT
+            HOUR(timestamp) AS hour_slot,
+            COUNT(*) AS message_count
+        FROM messages
+        WHERE group_id = %s
+          AND timestamp >= DATE_SUB(NOW(), INTERVAL %s DAY)
+        GROUP BY HOUR(timestamp)
+        ORDER BY hour_slot
+        """
+        rows = self.execute_query(sql, (group_id, days)) or []
+        return [
+            {
+                "hour": int(row.get("hour_slot") or 0),
+                "message_count": int(row.get("message_count") or 0),
+            }
+            for row in rows
+        ]
+
+    def get_group_last_message(self, group_id: str) -> Optional[Dict]:
+        """获取群最后一条消息信息"""
+        sql = """
+        SELECT sender, content, message_type, timestamp
+        FROM messages
+        WHERE group_id = %s
+        ORDER BY timestamp DESC
+        LIMIT 1
+        """
+        row = self.execute_query(sql, (group_id,), fetch_one=True)
+        if row and isinstance(row.get("timestamp"), datetime):
+            row["timestamp"] = row["timestamp"].strftime("%Y-%m-%d %H:%M:%S")
+        return row
+
     def get_messages_by_filter(self, group_id=None, start_date=None, end_date=None,
                                search_text=None, page=1, page_size=20) -> Dict:
         """按条件筛选消息并支持分页和模糊搜索

db/stats_db.py

@@ -555,4 +555,56 @@ class StatsDBOperator(BaseDBOperator):
                 })
             current_date += timedelta(days=1)
         
-        return trend_data
+        return trend_data
+
+    def get_group_plugin_stats(self, group_id: str, days: int = 30, limit: int = 10) -> List[Dict]:
+        """获取指定群的插件调用统计"""
+        sql = """
+        SELECT
+            plugin_name,
+            command,
+            SUM(total_calls) AS total_calls,
+            SUM(success_calls) AS success_calls,
+            SUM(failed_calls) AS failed_calls,
+            MAX(unique_users) AS unique_users,
+            MIN(first_used_at) AS first_used_at,
+            MAX(last_used_at) AS last_used_at
+        FROM t_group_stats
+        WHERE group_id = %s
+          AND last_used_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
+        GROUP BY plugin_name, command
+        ORDER BY total_calls DESC, last_used_at DESC
+        LIMIT %s
+        """
+        rows = self.execute_query(sql, (group_id, days, limit)) or []
+        for row in rows:
+            for key in ("first_used_at", "last_used_at"):
+                dt = row.get(key)
+                if isinstance(dt, datetime):
+                    row[key] = dt.strftime("%Y-%m-%d %H:%M:%S")
+        return rows
+
+    def get_group_plugin_summary(self, group_id: str, days: int = 30) -> Dict:
+        """获取指定群的插件调用摘要"""
+        sql = """
+        SELECT
+            SUM(total_calls) AS total_calls,
+            SUM(success_calls) AS success_calls,
+            SUM(failed_calls) AS failed_calls,
+            COUNT(DISTINCT plugin_name) AS plugin_count,
+            MAX(last_used_at) AS last_used_at
+        FROM t_group_stats
+        WHERE group_id = %s
+          AND last_used_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
+        """
+        result = self.execute_query(sql, (group_id, days), fetch_one=True) or {}
+        dt = result.get("last_used_at")
+        if isinstance(dt, datetime):
+            result["last_used_at"] = dt.strftime("%Y-%m-%d %H:%M:%S")
+        return {
+            "total_calls": int(result.get("total_calls") or 0),
+            "success_calls": int(result.get("success_calls") or 0),
+            "failed_calls": int(result.get("failed_calls") or 0),
+            "plugin_count": int(result.get("plugin_count") or 0),
+            "last_used_at": result.get("last_used_at") or "",
+        }