需求：1.加入了用户积分表；2.加入了指令积分扣除功能；3.加入了积分获得与扣除注解。

utils/ai/game_chatgpt_qa.py

@@ -0,0 +1,102 @@
+import requests
+import json
+import random
+
+# 解析JSON
+def extract_content(data_string):
+    try:
+        data = json.loads(data_string)
+        # 提取content字段
+        content = data["choices"][0]["message"].get("content", "")
+        return content
+    except json.JSONDecodeError:
+        print("Invalid JSON")
+        return None
+
+
+def message_task_json(prompt, content):
+    # 设置Authorization和URL
+    authorization = "Bearer b8586595-eb81-483d-8e91-a35cc789729e"  # 请替换为真实的Authorization token
+    url = 'https://ark.cn-beijing.volces.com/api/v3/chat/completions'
+
+    data = {
+        # "stream": True,
+        "model": "doubao-1-5-lite-32k-250115",
+        "messages": [
+            {
+                "role": "system",
+                "content": f"{prompt}"
+            },
+            {
+                "role": "user",
+                "content": f"{content}"
+            }
+
+        ]
+    }
+
+    # 设置请求头
+    headers = {
+        "Content-Type": "application/json; charset=utf-8",
+        "Authorization": authorization
+    }
+
+    # 发送POST请求
+    response = requests.post(url, headers=headers, data=json.dumps(data), )
+    response.encoding = 'utf-8'
+
+    # 输出响应内容
+    print(response.status_code)
+    print(response.text)
+    return json.loads(extract_content(response.text))
+
+
+def game_question_json(question):
+    fields = [
+        "近现代史", "战争与政治", "文化遗产与考古学", "进化论", "动植物学", "基因与遗传学",
+        "生态学", "有机化学", "无机化学", "生物化学", "环境化学", "人文地理", "自然地理",
+        "地质学", "气候变化", "古典文学", "现代文学", "小说与戏剧", "诗歌与散文", "美术",
+        "音乐", "戏剧与舞蹈", "电影与媒体", "西方哲学", "东方哲学", "道德与伦理学", "逻辑与认识论",
+        "人类学", "心理学", "政治学", "经济学", "编程语言", "人工智能", "数据科学", "网络与安全",
+        "机械工程", "电气工程", "化学工程", "土木工程", "解剖学", "生理学", "临床医学", "药学与护理",
+        "球类运动", "奥林匹克运动", "运动心理学", "健身与营养", "世界宗教", "神话与民间故事",
+        "宗教哲学", "语法与词汇", "语言习得", "方言与语言变异", "宏观经济学", "微观经济学",
+        "国际贸易", "金融与投资", "民法与刑法", "国际法", "知识产权法", "环境法"
+    ]
+
+    # 随机选择一个领域
+    selected_field = random.choice(fields)
+
+    # 输出随机选择的领域
+    print(f"随机选择的领域是：{selected_field}")
+    question = question + f"随机选择的领域是：{selected_field}"
+    prompt = """
+    请根据以下要求，随机生成一个问题，确保每次提问涉及不同领域，且不重复：
+    1. 每个问题应该覆盖以下任意领域：近现代史、战争与政治、文化遗产与考古学、进化论、动植物学、基因与遗传学、生态学、有机化学、无机化学、生物化学、环境化学、人文地理、自然地理、地质学、气候变化、古典文学、现代文学、小说与戏剧、诗歌与散文、美术、音乐、戏剧与舞蹈、电影与媒体、西方哲学、东方哲学、道德与伦理学、逻辑与认识论、人类学、心理学、政治学、经济学、编程语言、人工智能、数据科学、网络与安全、机械工程、电气工程、化学工程、土木工程、解剖学、生理学、临床医学、药学与护理、球类运动、奥林匹克运动、运动心理学、健身与营养、世界宗教、神话与民间故事、宗教哲学、语法与词汇、语言习得、方言与语言变异、宏观经济学、微观经济学、国际贸易、金融与投资、民法与刑法、国际法、知识产权法、环境法。
+    2. 问题应简洁，具有一定难度，易于理解且充满乐趣，适合百科类知识问答。
+    3. 每个问题应独立，且问题之间无连贯性或延续性。
+    4. 避免重复提问或产生相似的问题，确保问题新颖。
+    5. 对于每个问题，提供难度评分（1-10分）。在问题答对后，用户可以根据难度给出相应的分数。
+    6. 答案控制长度在25字以内。
+    
+    输出格式要求如下（仅返回JSON格式，确保不添加多余的符号）：
+    {   
+        "category":"人文地理"
+        "question": "哪个国家最早将玫瑰与爱情联系起来？",
+        "score": "1",
+        "answer": "波斯",
+        "description": "波斯文化中，玫瑰被广泛认为是象征爱情的花卉，其象征意义逐渐传入西方。"
+    }
+    """
+
+    return message_task_json(prompt, question)
+
+
+def game_answer_json(answer):
+    prompt = "你是一个益智百科问答大师，可以根据用户回答的答案进行判断，并且对问题(question)答案(answer)进行打分,打分时请参考最高分要求(top_score)，告知用户能获得多少分，请在description中描述打分理由，请只返回JSON格式的内容：格式要求如下(请不要加上markdown 的符号)：{\"question\": \"哪个国家最早将玫瑰与爱情联系起来？\", \"score\":\"1\",  \"answer\": \"波斯\",\"description\":\"描述问题答案的原因\"}"
+    return message_task_json(prompt, answer)
+
+
+if __name__ == '__main__':
+    print(game_question_json('请出题！'))
+    # print(game_answer_json('question:哪个国家的节日与裸体狂欢有关？,answer:古罗马,top_score:3'))

utils/decorator/plugin_decorators.py

@@ -0,0 +1,120 @@
+import functools
+import time
+import traceback
+import logging
+from datetime import datetime
+from typing import Callable, Dict, Any, Tuple
+
+from db.stats_db import StatsDBOperator
+from db.connection import DBConnectionManager
+
+
+def plugin_stats_decorator(plugin_name: str) -> Callable:
+    """插件统计装饰器
+    
+    Args:
+        plugin_name: 插件名称
+        
+    Returns:
+        装饰器函数
+    """
+    # 获取日志记录器
+    logger = logging.getLogger(f"StatsCollector.{plugin_name}")
+    logger.debug(f"为插件 '{plugin_name}' 应用统计装饰器")
+    
+    def decorator(func: Callable) -> Callable:
+        logger.debug(f"装饰 '{plugin_name}' 的 {func.__name__} 方法")
+        
+        @functools.wraps(func)
+        def wrapper(self, message: Dict[str, Any]) -> Tuple[bool, str]:
+            # 获取数据库连接
+            try:
+                logger.debug(f"[{plugin_name}] 开始处理消息")
+                db_manager = DBConnectionManager.get_instance()
+                stats_db = StatsDBOperator(db_manager)
+                
+                # 提取消息信息
+                content = message.get("content", "")
+                sender = message.get("sender", "")
+                roomid = message.get("roomid", "")
+                
+                # 提取指令部分（假设指令是第一个单词或空格前的部分）
+                command = content.strip().split(' ')[0] if content else ""
+                
+                logger.debug(f"[{plugin_name}] 消息内容: '{content}', 指令: '{command}', 发送者: {sender}, 群ID: {roomid}")
+                
+                # 记录开始时间
+                start_time = time.time()
+                logger.debug(f"[{plugin_name}] 开始执行时间: {datetime.fromtimestamp(start_time).strftime('%Y-%m-%d %H:%M:%S.%f')}")
+                
+                try:
+                    # 调用原始方法
+                    logger.debug(f"[{plugin_name}] 调用原始方法 {func.__name__}")
+                    success, response = func(self, message)
+                    
+                    # 计算执行时间(毫秒)
+                    end_time = time.time()
+                    process_time_ms = (end_time - start_time) * 1000
+                    logger.debug(f"[{plugin_name}] 执行完成，耗时: {process_time_ms:.2f}ms, 结果: {success}, 响应: {response}")
+                    
+                    # 记录插件调用
+                    logger.debug(f"[{plugin_name}] 记录插件调用统计")
+                    stats_db.record_plugin_call(
+                        plugin_name=plugin_name,
+                        command=command,  # 使用提取的指令而不是完整内容
+                        user_id=sender,
+                        group_id=roomid,
+                        success=success,
+                        process_time_ms=process_time_ms
+                    )
+                    logger.info(f"[{plugin_name}] 成功记录插件调用: {command}, 耗时: {process_time_ms:.2f}ms")
+                    
+                    return success, response
+                except Exception as e:
+                    # 计算执行时间(毫秒)
+                    end_time = time.time()
+                    process_time_ms = (end_time - start_time) * 1000
+                    
+                    # 记录错误
+                    error_message = str(e)
+                    stack_trace = traceback.format_exc()
+                    logger.error(f"[{plugin_name}] 执行出错: {error_message}")
+                    logger.debug(f"[{plugin_name}] 错误堆栈: {stack_trace}")
+                    
+                    try:
+                        # 记录插件调用（失败）
+                        logger.debug(f"[{plugin_name}] 记录插件调用失败统计")
+                        stats_db.record_plugin_call(
+                            plugin_name=plugin_name,
+                            command=command,  # 使用提取的指令而不是完整内容
+                            user_id=sender,
+                            group_id=roomid,
+                            success=False,
+                            process_time_ms=process_time_ms
+                        )
+                        
+                        # 记录错误详情
+                        logger.debug(f"[{plugin_name}] 记录错误详情")
+                        stats_db.record_error(
+                            plugin_name=plugin_name,
+                            command=command,  # 使用提取的指令而不是完整内容
+                            user_id=sender,
+                            group_id=roomid,
+                            error_message=error_message,
+                            stack_trace=stack_trace
+                        )
+                        logger.info(f"[{plugin_name}] 成功记录插件错误: {command}, 错误: {error_message}")
+                    except Exception as db_error:
+                        logger.error(f"[{plugin_name}] 记录插件统计数据失败: {db_error}")
+                    
+                    # 重新抛出异常，让上层处理
+                    raise
+            except Exception as outer_error:
+                logger.error(f"[{plugin_name}] 装饰器外层错误: {outer_error}")
+                logger.error(traceback.format_exc())
+                # 确保原始函数仍然被调用，即使装饰器出错
+                return func(self, message)
+        
+        return wrapper
+    
+    return decorator

utils/decorator/points_decorator.py

@@ -0,0 +1,164 @@
+import functools
+import time
+import traceback
+import logging
+from datetime import datetime
+from typing import Callable, Dict, Any, Tuple, Union
+
+from db.connection import DBConnectionManager
+from db.points_db import PointsDBOperator, PointSource
+
+
+def points_reward_decorator(points_calculator: Union[int, Callable], source_type: str = "other",
+                            description: str = None):
+    """积分奖励装饰器
+    
+    Args:
+        points_calculator: 积分数量或计算函数，如果是函数，接收(self, message, success, response)参数并返回积分数量
+        source_type: 积分来源类型 (checkin, game, other)
+        description: 积分奖励描述
+        
+    Returns:
+        装饰器函数
+    """
+
+    def decorator(func: Callable) -> Callable:
+        @functools.wraps(func)
+        def wrapper(self, message: Dict[str, Any]) -> Tuple[bool, str]:
+            # 调用原始方法
+            success, response = func(self, message)
+
+            # 如果原始方法执行成功，奖励积分
+            if success:
+                try:
+                    # 获取消息信息
+                    sender = message.get("sender", "")
+                    roomid = message.get("roomid", "")
+
+                    if sender and (roomid or sender):
+                        # 计算奖励积分数量
+                        if callable(points_calculator):
+                            # 如果是函数，调用函数计算积分
+                            points = points_calculator(self, message, success, response)
+                            if not points or points <= 0:
+                                # 如果计算结果为0或负数，不奖励积分
+                                return success, response
+                        else:
+                            # 如果是固定值，直接使用
+                            points = points_calculator
+
+                        # 获取积分来源类型
+                        source = PointSource.CHECKIN
+                        if source_type.lower() == "game":
+                            source = PointSource.GAME
+                        elif source_type.lower() != "checkin":
+                            source = PointSource.OTHER
+
+                        # 奖励积分
+                        db_manager = DBConnectionManager.get_instance()
+                        points_db = PointsDBOperator(db_manager)
+
+                        # 如果description是函数，调用函数获取描述
+                        desc = description
+                        if callable(description):
+                            desc = description(self, message, success, response, points)
+
+                        reward_success, reward_result = points_db.add_points(
+                            sender, roomid, points, source,
+                            desc or f"使用 {self.name if hasattr(self, 'name') else '功能'} 获得奖励"
+                        )
+
+                        logger = logging.getLogger(f"PointsReward.{self.name if hasattr(self, 'name') else 'Unknown'}")
+                        if reward_success:
+                            logger.info(f"用户 {sender} 获得 {points} 积分奖励")
+
+                            # 如果响应中没有提到积分，添加积分信息
+                            if "积分" not in response:
+                                response += f"\n\n🎁 恭喜获得 {points} 积分奖励！"
+                        else:
+                            logger.warning(f"用户 {sender} 积分奖励失败: {reward_result}")
+                except Exception as e:
+                    logger = logging.getLogger("PointsReward")
+                    logger.error(f"奖励积分失败: {e}")
+                    logger.error(traceback.format_exc())
+
+            return success, response
+
+        return wrapper
+
+    return decorator
+
+
+def plugin_points_cost(points: int, description: str = None):
+    """插件积分消费装饰器
+    
+    Args:
+        points: 消费积分数量
+        description: 积分消费描述
+        
+    Returns:
+        装饰器函数
+    """
+
+    def decorator(func: Callable) -> Callable:
+        @functools.wraps(func)
+        def wrapper(self, message: Dict[str, Any]) -> Tuple[bool, str]:
+            try:
+                # 获取消息信息
+                sender = message.get("sender", "")
+                roomid = message.get("roomid", "")
+
+                if not sender or not (roomid or sender):
+                    return func(self, message)
+
+                # 检查用户积分是否足够
+                db_manager = DBConnectionManager.get_instance()
+                points_db = PointsDBOperator(db_manager)
+
+                plugin_name = self.name if hasattr(self, 'name') else "未知插件"
+                logger = logging.getLogger(f"PointsCost.{plugin_name}")
+
+                user_points = points_db.get_user_points(sender, roomid)
+                if user_points["total_points"] < points:
+                    # 积分不足
+                    wcf = message.get("wcf")
+                    if wcf:
+                        wcf.send_text(
+                            f"❌ 积分不足，无法使用 {plugin_name} 功能\n"
+                            f"当前积分: {user_points['total_points']}\n"
+                            f"需要积分: {points}\n"
+                            f"还差 {points - user_points['total_points']} 积分",
+                            (roomid if roomid else sender), sender
+                        )
+                    logger.info(f"用户 {sender} 积分不足，无法使用功能")
+                    return False, "积分不足"
+
+                # 调用原始方法
+                success, response = func(self, message)
+
+                # 如果原始方法执行成功，扣除积分
+                if success:
+                    deduct_success, deduct_result = points_db.deduct_points(
+                        sender, roomid, points, PointSource.PLUGIN,
+                        description or f"使用 {plugin_name} 功能"
+                    )
+
+                    if deduct_success:
+                        logger.info(f"用户 {sender} 使用功能扣除 {points} 积分")
+
+                        # 如果响应中没有提到积分，添加积分信息
+                        if "积分" not in response:
+                            response += f"\n\n💰 已消费 {points} 积分"
+                    else:
+                        logger.warning(f"用户 {sender} 积分扣除失败: {deduct_result}")
+
+                return success, response
+            except Exception as e:
+                logger = logging.getLogger("PointsCost")
+                logger.error(f"积分消费失败: {e}")
+                logger.error(traceback.format_exc())
+                return func(self, message)
+
+        return wrapper
+
+    return decorator

utils/robot_cmd/robot_command.py

@@ -0,0 +1,381 @@
+# 群清单管理
+# 群功能管理
+# 0.加入或者关闭群机器人 #启用群机器人 #关闭群机器人
+# 1.每日新闻自动播报 #启用每日新闻播报 #关闭每日新闻播报
+# 2.每日群发言总结 #启用群发言 #关闭群发言
+# 3.群AI能力 #启用群AI #关闭群AI
+# 4.群总结能力 #启用群总结 #关闭群总结
+# 5.sehuatang PDF能力 #启用pdf #关闭pdf
+from typing import List
+
+import redis
+import json
+from enum import Enum
+
+# 连接到本地 Redis 服务
+r = redis.StrictRedis(host='192.168.2.32', port=6379, db=0, decode_responses=True)
+
+
+class PermissionStatus(Enum):
+    """权限状态枚举"""
+    ENABLED = "enabled"
+    DISABLED = "disabled"
+
+
+class Feature(Enum):
+    """功能权限枚举，带序号"""
+    ROBOT = 1, "群机器人"
+    DAILY_NEWS = 2, "每日新闻自动播报"
+    DAILY_SUMMARY = 3, "每日群发言总结"
+    AI_CAPABILITY = 4, "群AI能力 [ai, dify, 聊天, AI] "
+    SUMMARY_CAPABILITY = 5, "群总结能力 [#总结]"
+    PDF_CAPABILITY = 6, "sehuatang PDF能力"
+    EPIC = 7, "EPIC自动播报"  # 新增的功能
+    PIC = 8, "图来能力 [图来, 秀人]"
+    TASK_GAME = 9, "百科答题游戏 [/t]"
+    MUSIC = 10, "点歌功能 [点歌, 音乐, 音乐点播, 点播音乐, 音乐点歌]"
+    SIGNIN = 11, "签到功能 [签到, 每日签到, qd, Qd, QD, 上班, 牛马]"
+    POINT_TRADE = 12, "积分赠送功能 [积分交易, 积分转账, 转账积分, 积分赠送, 赠送积分, 积分转移]"
+    BEAUTY_LEG = 13, "腿来能力 [美腿, 腿来]"
+    VIDEO = 14, "视频点播功能 [黑丝视频, 黑丝, 来个黑丝,搞个黑丝]"
+    VIDEO_MAN = 15, "视频肌肉男点播功能 [猛男, 肌肉, 帅哥]"
+    GROUP_ADD = 16, "加群提醒功能"
+    DOUYIN_PARSER = 17, "抖音链接转视频功能"
+    GROUP_MEMBER_CHANGE = 18, "群成员变更提醒功能"
+
+    def __new__(cls, value, description):
+        obj = object.__new__(cls)
+        obj._value_ = value
+        obj.description = description  # 添加描述
+        return obj
+
+    def __str__(self):
+        return self.description
+
+
+class GroupBotManager:
+    """群机器人管理，支持本地缓存"""
+
+    # 本地缓存作为类级别静态属性
+    local_cache = {
+        "group_permissions": {},  # 用于缓存群组功能权限
+        "group_list": set()  # 用于缓存 group:list
+    }
+
+    @staticmethod
+    def display_menu_status(group_id):
+        """显示所有功能列表及其在指定群组中的当前状态，带emoji"""
+        menu = []
+        for feature in Feature:
+            status = GroupBotManager.get_group_permission(group_id, feature)
+            status_emoji = "✅" if status == PermissionStatus.ENABLED else "❌"
+            status_str = "启用" if status == PermissionStatus.ENABLED else "关闭"
+            menu.append(f"{status_emoji} {status_str}-{feature.value}-{feature.description}")
+        return "\n".join(menu)
+
+    @staticmethod
+    def load_local_cache():
+        """从 Redis 加载数据到本地缓存"""
+        group_list = r.smembers("group:list")
+        GroupBotManager.local_cache["group_list"] = set(group_list)
+
+        # 加载群组权限
+        for group_id in GroupBotManager.local_cache["group_list"]:
+            key = f'group:{group_id}:permissions'
+            GroupBotManager.local_cache["group_permissions"][group_id] = {}
+            for feature in Feature:
+                status_value = r.hget(key, feature.name)
+                if status_value:
+                    GroupBotManager.local_cache["group_permissions"][group_id][feature] = PermissionStatus(status_value)
+                else:
+                    GroupBotManager.local_cache["group_permissions"][group_id][feature] = PermissionStatus.DISABLED
+
+    @staticmethod
+    def save_to_redis():
+        """将本地缓存保存回 Redis"""
+        # 保存 group:list 到 Redis
+        r.sadd("group:list", *GroupBotManager.local_cache["group_list"])
+
+        # 保存每个群组的权限到 Redis
+        for group_id, permissions in GroupBotManager.local_cache["group_permissions"].items():
+            key = f'group:{group_id}:permissions'
+            for feature, status in permissions.items():
+                r.hset(key, feature.name, status.value)
+
+    @staticmethod
+    def set_group_permission(group_id, feature: Feature, status: PermissionStatus):
+        """设置群组功能权限并更新本地缓存"""
+        # 更新本地缓存
+        if group_id not in GroupBotManager.local_cache["group_permissions"]:
+            GroupBotManager.local_cache["group_permissions"][group_id] = {}
+
+        GroupBotManager.local_cache["group_permissions"][group_id][feature] = status
+
+        # 同步到 Redis
+        key = f'group:{group_id}:permissions'
+        r.hset(key, feature.name, status.value)
+
+    @staticmethod
+    def get_group_permission(group_id, feature: Feature):
+        """获取群组某个功能的权限状态"""
+        # 先从本地缓存获取
+        if group_id in GroupBotManager.local_cache["group_permissions"]:
+            return GroupBotManager.local_cache["group_permissions"][group_id].get(feature, PermissionStatus.DISABLED)
+        else:
+            return PermissionStatus.DISABLED
+
+    @staticmethod
+    def check_permission(group_id, feature: Feature):
+        """检查某个功能是否启用，若未启用则返回提示信息"""
+        status = GroupBotManager.get_group_permission(group_id, feature)
+        if status == PermissionStatus.DISABLED:
+            return f"该功能未启用，请开启 {feature.description}。"
+        return None  # 如果已启用，则返回 None，不做处理
+
+    @staticmethod
+    def handle_command(group_id, command_str):
+        """统一处理群功能指令"""
+        print(f"PermissionStatus handle_command command_str: {command_str}")
+        # 命令解析
+        command_parts = command_str.strip().split("-")
+
+        # 如果是MENU指令，返回功能列表
+        if command_str.strip().upper() == "菜单":
+            return f"群ID：{group_id} \n {GroupBotManager.display_menu_status(group_id)}"
+
+        # 如果是MENU-STATUS指令，返回功能列表及其状态
+        if command_str.strip().upper() == "菜单状态":
+            return f"群ID：{group_id} \n {GroupBotManager.display_menu_status(group_id)}"
+
+        # 如果是GROUP_LIST指令，返回 group:list 清单
+        if command_str.strip().upper() == "群列表":
+            return GroupBotManager.get_group_list()
+
+        # 如果是清除群指令
+        if command_str.strip().startswith("清除群-"):
+            target_group_id = command_str.strip().split("-")[1]
+            return GroupBotManager.remove_group(target_group_id)
+
+        if len(command_parts) < 2:
+            return None
+
+        feature_str = command_parts[0]
+        action = command_parts[1]
+
+        # 如果第一个参数是序号，则转化为对应的功能
+        if feature_str.isdigit():
+            feature_num = int(feature_str)
+            try:
+                feature = Feature(feature_num)  # 使用枚举序号查找功能
+            except ValueError:
+                return "无效的功能序号"
+        else:
+            try:
+                feature = Feature[feature_str]  # 通过枚举名称获取功能枚举
+            except KeyError:
+                return "无效功能名称"
+
+        # 处理群机器人的启用和关闭（特别操作：更新 group:list）
+        if feature == Feature.ROBOT:
+            if action == "启用":
+                GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.ENABLED)
+                # 启用群机器人时，将 group_id 加入 group:list
+                GroupBotManager.local_cache["group_list"].add(group_id)
+                # 同步到 Redis
+                r.sadd("group:list", group_id)
+                return f"BOT已启用，群组 {group_id} 已加入"
+            elif action == "关闭":
+                GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.DISABLED)
+                # 关闭群机器人时，从 group:list 中删除 group_id
+                GroupBotManager.local_cache["group_list"].remove(group_id)
+                # 同步到 Redis
+                r.srem("group:list", group_id)
+                return f"BOT已关闭，群组 {group_id} 已移除"
+            else:
+                return "无效操作，仅支持启用或关闭"
+
+        # 先检查群机器人权限
+        robot_status = GroupBotManager.get_group_permission(group_id, Feature.ROBOT)
+        if robot_status != PermissionStatus.ENABLED and feature != Feature.ROBOT:
+            return "群机器人未启用，无法执行其他功能操作"
+
+        # 根据不同的操作启用或禁用功能
+        if action == "启用":
+            GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.ENABLED)
+            return f"群ID：{group_id} \n {feature.description} 已启用"
+        elif action == "关闭":
+            GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.DISABLED)
+            return f"群ID：{group_id} \n {feature.description} 已关闭"
+        else:
+            return "无效操作，仅支持启用或关闭"
+
+    @staticmethod
+    def list_group_permissions(group_id):
+        """列出群组所有功能及其状态"""
+        permissions = {}
+        for feature in Feature:
+            status = GroupBotManager.get_group_permission(group_id, feature)
+            permissions[feature] = status if status else PermissionStatus.DISABLED  # 默认为禁用状态
+        return permissions
+
+    @staticmethod
+    def display_menu():
+        """显示所有功能列表及其当前状态"""
+        menu = []
+        for feature in Feature:
+            menu.append(f"{feature.value}. {feature.description}")
+        return "\n".join(menu)
+
+    @staticmethod
+    def get_enabled_features(group_id):
+        """获取某个群已启用的功能列表及其描述，并返回格式化的字符串
+        只返回描述中包含指令(方括号[])的功能
+        
+        Args:
+            group_id: 群ID
+            
+        Returns:
+            str: 格式化的已启用功能列表字符串
+        """
+        enabled_features = []
+        
+        # 检查群是否在列表中
+        if group_id not in GroupBotManager.local_cache["group_list"]:
+            return "该群未启用机器人功能"
+            
+        # 遍历所有功能，检查哪些已启用且包含指令
+        for feature in Feature:
+            status = GroupBotManager.get_group_permission(group_id, feature)
+            # 只添加已启用且描述中包含方括号的功能
+            if status == PermissionStatus.ENABLED and "[" in feature.description and "]" in feature.description:
+                enabled_features.append({
+                    "id": feature.value,
+                    "name": feature.name,
+                    "description": feature.description
+                })
+        
+        # 如果没有启用任何带指令的功能
+        if not enabled_features:
+            return "该群未启用任何带指令的功能"
+            
+        # 构建格式化的字符串
+        result = f"群功能菜单：\n"
+        for feature in enabled_features:
+            result += f"{feature['id']}.{feature['description']}\n"
+                
+        return result
+        
+    @staticmethod
+    def get_group_list():
+        """返回所有启用了群机器人的群组清单，格式为集合"""
+        return list(GroupBotManager.local_cache["group_list"])
+
+    @staticmethod
+    def remove_group(group_id):
+        """一键清除某个群的所有设置，用于退群或关闭群时处理
+        
+        Args:
+            group_id: 群ID
+            
+        Returns:
+            str: 操作结果信息
+        """
+        # 检查群是否在列表中
+        if group_id not in GroupBotManager.local_cache["group_list"]:
+            return f"群 {group_id} 不在机器人管理列表中"
+
+        # 从本地缓存中移除群组
+        GroupBotManager.local_cache["group_list"].remove(group_id)
+        if group_id in GroupBotManager.local_cache["group_permissions"]:
+            del GroupBotManager.local_cache["group_permissions"][group_id]
+
+        # 从Redis中移除群组
+        r.srem("group:list", group_id)
+        r.delete(f'group:{group_id}:permissions')
+
+        return f"已成功清除群 {group_id} 的所有设置"
+
+    @staticmethod
+    def get_admin_list() -> List[str]:
+        """获取管理员列表
+        
+        返回系统管理员的微信ID列表
+        """
+        # 从配置文件中获取管理员列表
+        config_admin_list = []  # self.config.get("admin_list", [])
+
+        # 手动添加的管理员ID列表
+        manual_admin_list = [
+            "Jyunere",  # 示例ID，请替换为实际的微信ID
+            "wxid_abcdef",  # 示例ID，请替换为实际的微信ID
+            "filehelper"  # 文件传输助手，方便自己测试
+        ]
+
+        # 合并所有管理员列表并去重
+        all_admin_list = list(set(config_admin_list + manual_admin_list))
+
+        return all_admin_list
+
+
+# 示例命令
+def simulate_commands():
+    # 加载本地缓存
+    GroupBotManager.load_local_cache()
+
+    group_id = "49571962306@chatroom"
+
+    print(GroupBotManager.get_group_permission(group_id, Feature.AI_CAPABILITY) == PermissionStatus.DISABLED)
+
+    print(GroupBotManager.get_group_permission(group_id, Feature.SUMMARY_CAPABILITY) == PermissionStatus.ENABLED)
+    # # 启用群机器人
+    # print(GroupBoManager.handle_command(group_id, "ROBOT-启用"))
+    #
+    # # 启用每日新闻自动播报
+    # print(GroupBotManager.handle_command(group_id, "2-启用"))  # 使用序号启用每日新闻自动播报
+    #
+    # # 关闭群AI能力
+    # print(GroupBotManager.handle_command(group_id, "4-关闭"))  # 使用序号关闭群AI能力
+    #
+    # # 启用群总结能力
+    # print(GroupBotManager.handle_command(group_id, "5-启用"))  # 使用序号启用群总结能力
+    #
+    # # 关闭Sehuatang PDF能力
+    # print(GroupBotManager.handle_command(group_id, "6-关闭"))  # 使用序号关闭Sehuatang PDF能力
+    #
+    # # 启用EPIC自动播报
+    # print(GroupBotManager.handle_command(group_id, "7-启用"))  # 使用序号启用EPIC自动播报
+    #
+    # # 查看当前群组的功能权限
+    # print(GroupBotManager.get_group_permission(group_id, Feature.ROBOT))
+    # print(GroupBotManager.get_group_permission(group_id, Feature.DAILY_NEWS))
+    # print(GroupBotManager.get_group_permission(group_id, Feature.AI_CAPABILITY))
+    # print(GroupBotManager.get_group_permission(group_id, Feature.SUMMARY_CAPABILITY))
+    # print(GroupBotManager.get_group_permission(group_id, Feature.PDF_CAPABILITY))
+    # print(GroupBotManager.get_group_permission(group_id, Feature.EPIC))
+    #
+    # # 查看群组所有功能和状态
+    # permissions = GroupBotManager.list_group_permissions(group_id)
+    # for feature, status in permissions.items():
+    #     print(f"{feature.description} (序号: {feature.value}): {status.value}")
+    #
+    # # 查看 group:list 中的群组
+    # print("当前启用群机器人的群组：", GroupBotManager.get_group_list())
+    #
+    # # 查看菜单功能列表
+    # print("功能列表：")
+    # print(GroupBotManager.handle_command(group_id, "MENU"))
+    #
+    # # 查看 group:list 清单
+    # print("群组清单：")
+    # print(GroupBotManager.handle_command(group_id, "GROUP_LIST"))
+    #
+    # # 更新缓存
+    # print(GroupBotManager.handle_command(group_id, "UPDATE"))
+    #
+    # # 保存到 Redis
+    # GroupBotManager.save_to_redis()
+
+
+if __name__ == '__main__':
+    # 执行模拟命令
+    simulate_commands()

utils/wechat/message_to_db.py

@@ -0,0 +1,343 @@
+from datetime import datetime, timedelta
+import xml.etree.ElementTree as ET
+import logging
+import concurrent.futures  # 添加线程池支持
+import os
+from wcferry import WxMsg, Wcf
+
+from db.connection import DBConnectionManager
+from db.message_storage import MessageStorageDB
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger("MessageStorage")
+
+
+class MessageStorage:
+
+    def __init__(self, wcf: Wcf = None):
+        # 获取数据库连接管理器的单例
+        self.db_manager = DBConnectionManager.get_instance()
+        self.message_db = MessageStorageDB(self.db_manager)
+        # 初始化本地缓存字典，使用 group_id 作为键
+        self.local_membercounts = {}
+        self.local_members = {}
+        # 创建线程池，用于异步存储消息
+        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+        # 用于跟踪异步任务的列表
+        self.pending_tasks = []
+
+        # 保存WCF实例，用于图片处理
+        self.wcf = wcf
+
+        # 图片处理相关初始化
+        self.image_executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)  # 专用于图片处理的线程池
+        self.image_tasks = []
+        self.image_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "static", "images")
+        # 确保图片存储目录存在
+        if not os.path.exists(self.image_dir):
+            os.makedirs(self.image_dir, exist_ok=True)
+        logger.info(f"图片存储目录: {self.image_dir}")
+
+    def process_message(self, message: WxMsg):
+        # 示例message字符串
+        current_date = datetime.now().strftime('%Y-%m-%d')
+        # 生成Redis key
+        key = f"{message.roomid}:{message.sender}:{current_date}:count"
+        # 获取 Redis 连接
+        redis_conn = self.db_manager.get_redis_connection()
+        # 使用Redis哈希（或字符串）增加发言次数
+        redis_conn.hincrby(key, 'count', 1)  # 这里使用哈希，但也可以考虑用字符串的INCR操作
+        # 设置时效为48小时
+        redis_conn.expire(key, 86400 * 2)
+        # 或者使用字符串：r.incr(key)  # 如果只存储一个整数值，字符串类型可能更简单
+
+    def archive_message(self, msg: WxMsg):
+        """异步存档消息，防止堵塞主线程"""
+        # 提交任务到线程池
+        future = self.executor.submit(self._archive_message_task, msg)
+        # 可选：添加回调函数处理完成后的操作
+        future.add_done_callback(self._archive_callback)
+        # 将任务添加到待处理列表
+        self.pending_tasks.append(future)
+        # 清理已完成的任务
+        self._cleanup_completed_tasks()
+
+    def _archive_message_task(self, msg: WxMsg):
+        """实际执行消息存档的任务函数"""
+        try:
+            # 使用 MessageStorageDB 类存档消息
+            result = self.message_db.archive_message(msg)
+            return {
+                'success': result,
+                'roomid': msg.roomid,
+                'sender': msg.sender,
+                'content': msg.content,  # 添加消息内容
+                'message_id': msg.id  # 添加消息ID
+            }
+        except Exception as e:
+            logger.error(f"存档消息出错: {e}")
+            return {
+                'success': False,
+                'roomid': msg.roomid,
+                'sender': msg.sender,
+                'content': msg.content,  # 添加消息内容
+                'message_id': msg.id,  # 添加消息ID
+                'error': str(e)
+            }
+
+    def process_image(self, msg: WxMsg):
+        """异步处理图片消息，与消息存档分离"""
+        if msg.type != 3 or not self.wcf:  # 不是图片消息或没有WCF实例
+            return False
+
+        # 提交任务到图片处理线程池
+        future = self.image_executor.submit(self._process_image_task, msg)
+        # 添加回调函数
+        future.add_done_callback(self._process_image_callback)
+        # 将任务添加到待处理列表
+        self.image_tasks.append(future)
+        # 清理已完成的任务
+        self._cleanup_completed_tasks()
+        return True
+
+    def _process_image_task(self, msg: WxMsg):
+        """实际执行图片处理的任务函数"""
+        try:
+            # 使用wcf下载图片，确保图片存在
+            if self.wcf and msg.id:
+                # 创建按群ID或个人wxid分割的目录
+                target_dir = os.path.join(self.image_dir, msg.roomid if msg.roomid else msg.sender)
+                # 确保目标目录存在
+                if not os.path.exists(target_dir):
+                    os.makedirs(target_dir, exist_ok=True)
+
+                # 尝试使用wcf下载图片到分组后的目录
+                download_path = self.wcf.download_image(msg.id, msg.extra, target_dir)
+                if download_path:
+                    logger.info(f"使用wcf下载图片成功: {msg.id} -> {download_path}")
+
+                    # 直接使用下载后的路径更新数据库
+                    self.message_db.update_message_image_path(msg.id, download_path)
+
+                    return {
+                        'success': True,
+                        'message_id': msg.id,
+                        'roomid': msg.roomid,
+                        'sender': msg.sender,
+                        'file_path': download_path
+                    }
+                else:
+                    return {
+                        'success': False,
+                        'message_id': msg.id,
+                        'roomid': msg.roomid,
+                        'sender': msg.sender,
+                        'error': "图片下载失败"
+                    }
+            else:
+                return {
+                    'success': False,
+                    'message_id': msg.id,
+                    'roomid': msg.roomid,
+                    'sender': msg.sender,
+                    'error': "WCF实例不存在或消息ID无效"
+                }
+        except Exception as e:
+            logger.error(f"图片处理出错: {msg.id}, 错误: {e}")
+            return {
+                'success': False,
+                'message_id': msg.id,
+                'roomid': msg.roomid,
+                'sender': msg.sender,
+                'error': str(e)
+            }
+
+    def _process_image_callback(self, future):
+        """处理异步图片处理任务完成后的回调"""
+        try:
+            result = future.result()
+            if result['success']:
+                skipped_info = " (已存在)" if result.get('skipped') else ""
+                logger.info(f"图片处理成功{skipped_info}: {result['roomid']}:{result['sender']}:{result['message_id']}")
+            else:
+                error_msg = result.get('error', '未知错误')
+                logger.error(
+                    f"图片处理失败: {result.get('roomid', '')}:{result.get('sender', '')}:{result.get('message_id', '')} - {error_msg}")
+        except Exception as e:
+            logger.error(f"处理图片回调时出错: {e}")
+
+    def _archive_callback(self, future):
+        """处理异步存档任务完成后的回调"""
+        try:
+            result = future.result()
+            if result['success']:
+                # 修改日志输出，包含消息内容
+                compressed = result['content'].replace('\n', '').replace('\r', '')
+                logger.info(f"消息存档成功: {result['roomid']}:{result['sender']}: {compressed}")
+            else:
+                error_msg = result.get('error', '未知错误')
+                logger.error(f"消息存档失败: {result['roomid']}:{result['sender']} - {error_msg}")
+        except Exception as e:
+            logger.error(f"处理存档回调时出错: {e}")
+
+    def _cleanup_completed_tasks(self):
+        """清理已完成的任务，防止内存泄漏"""
+        # 只有当任务数量超过阈值时才进行清理，减少频繁操作
+        if len(self.pending_tasks) > 20:
+            # 过滤出已完成的任务
+            completed_tasks = [task for task in self.pending_tasks if task.done()]
+            # 从待处理列表中移除已完成的任务
+            for task in completed_tasks:
+                self.pending_tasks.remove(task)
+
+            # 如果待处理任务过多，记录警告日志
+            if len(self.pending_tasks) > 100:
+                logger.warning(f"待处理的存档任务数量过多: {len(self.pending_tasks)}")
+
+        # 只有当图片任务数量超过阈值时才进行清理
+        if len(self.image_tasks) > 10:
+            # 清理已完成的图片处理任务
+            completed_image_tasks = [task for task in self.image_tasks if task.done()]
+            for task in completed_image_tasks:
+                self.image_tasks.remove(task)
+
+            # 如果待处理任务过多，记录警告日志
+            if len(self.image_tasks) > 50:
+                logger.warning(f"待处理的图片处理任务数量过多: {len(self.image_tasks)}")
+
+    def write_to_db(self):
+        """从Redis读取发言统计数据并写入数据库"""
+        # 获取Redis连接
+        redis_conn = self.db_manager.get_redis_connection()
+
+        # 获取当前日期的前一天
+        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
+
+        # 遍历Redis中所有与昨天日期相关的key，并写入数据库
+        for key_item in redis_conn.keys(f"*:*:{yesterday}:count"):
+            # 检查key是否为字节类型，如果是则解码
+            key = key_item.decode('utf-8') if isinstance(key_item, bytes) else key_item
+            parts = key.split(':')
+            group_id, wx_id, _date = parts[0], parts[1], parts[2]  # _date应该是yesterday
+
+            # 获取计数值
+            count_bytes = redis_conn.hget(key, 'count')
+            count = int(count_bytes) if isinstance(count_bytes, bytes) else int(count_bytes) if count_bytes else 0
+
+            # 使用MessageStorageDB插入数据
+            try:
+                result = self.message_db.insert_speech_count(group_id, wx_id, yesterday, count)
+                if result:
+                    logging.info(f"成功写入发言统计: {group_id}, {wx_id}, {yesterday}, {count}")
+                else:
+                    logging.error(f"写入发言统计失败: {group_id}, {wx_id}, {yesterday}, {count}")
+            except Exception as e:
+                logging.error(f"写入发言统计出错: {e}")
+
+    def generate_and_send_ranking(self, groupId, allContacts: dict):
+        """生成并发送群聊发言排名"""
+        try:
+            yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
+
+            # 使用数据库操作类获取排名数据
+            results = self.message_db.get_speech_ranking(yesterday, groupId, limit=20)
+
+            if not results:
+                logging.info(f"没有找到 {yesterday} 的群聊 {groupId} 发言记录")
+                return f"📊 {yesterday} 没有发言记录"
+
+            # 格式化输出字符串，添加emoji和美化格式
+            ranking_str = f"🏆 {yesterday} 发言排行榜 🏆\n"
+
+            # 为不同名次添加不同的奖杯和样式
+            for rank, result in enumerate(results, start=1):
+                username = result['wx_id']
+                speech_count = result['speech_count']
+                display_name = allContacts.get(username, username)
+
+                # 根据排名添加不同的emoji
+                if rank == 1:
+                    ranking_str += f"🥇🐲 {rank}.{display_name}: {speech_count}次 🔥\n"
+                elif rank == 2:
+                    ranking_str += f"🥈 {rank}.{display_name}: {speech_count}次 ✨\n"
+                elif rank == 3:
+                    ranking_str += f"🥉 {rank}.{display_name}: {speech_count}次 👏\n"
+                elif rank <= 10:
+                    ranking_str += f"🌟 {rank}.{display_name}: {speech_count}次\n"
+                else:
+                    ranking_str += f"👍 {rank}.{display_name}: {speech_count}次\n"
+
+            logging.info(f"成功生成 {yesterday} 的群聊 {groupId} 发言排名")
+            return ranking_str
+
+        except Exception as e:
+            logging.error(f"生成发言排名出错: {e}")
+            return f"❌ 生成发言排名出错: {e}"
+
+    def get_messages(self, group_id, all_contacts: dict):
+        try:
+            # 获取 Redis 连接
+            redis_conn = self.db_manager.get_redis_connection()
+
+            # 获取 redis 中的上次总结时间，本次从上次开始算，若没有，则从 8 小时之前开始计算
+            key = f"{group_id}:summary_time"
+            last_summary_time = redis_conn.get(key)
+            logger.info(f"上次总结时间: {last_summary_time}")
+
+            current_time = datetime.now()
+            current_date = current_time.strftime('%Y-%m-%d %H:%M:%S')
+
+            if not last_summary_time:
+                # 获取当前时间并计算 8 小时前的时间
+                eight_hours_ago = current_time - timedelta(hours=8)
+                last_summary_time = eight_hours_ago.strftime('%Y-%m-%d %H:%M:%S')
+            else:
+                # 如果 Redis 返回值为字节类型，转换为字符串
+                if isinstance(last_summary_time, bytes):
+                    last_summary_time = last_summary_time.decode('utf-8')
+
+                # 检查 redis 中的时间与当前时间差是否小于 3 小时
+                last_summary_time_obj = datetime.strptime(last_summary_time, '%Y-%m-%d %H:%M:%S')
+                time_diff = current_time - last_summary_time_obj
+
+                if time_diff < timedelta(hours=3):
+                    # 小于 3 小时，取 8 小时前
+                    last_summary_time = (current_time - timedelta(hours=8)).strftime('%Y-%m-%d %H:%M:%S')
+                elif time_diff > timedelta(days=1):
+                    # 大于 24 小时，取 10 小时前
+                    last_summary_time = (current_time - timedelta(hours=10)).strftime('%Y-%m-%d %H:%M:%S')
+
+            # 更新 Redis 存储的当前时间
+            redis_conn.set(key, current_date)
+
+            # 使用 MessageStorageDB 类获取最近消息
+            hours_ago = int(
+                (current_time - datetime.strptime(last_summary_time, '%Y-%m-%d %H:%M:%S')).total_seconds() / 3600) + 1
+            messages = self.message_db.get_recent_messages(group_id, hours_ago=hours_ago)
+            # 构建最终的结果字符串
+            result = []
+            for msg in messages:
+                timestamp, sender, content, message_type = msg['timestamp'], msg['sender'], msg['content'], msg[
+                    'message_type']
+                try:
+                    if message_type == 49:  # 应用消息类型
+                        # 其他类型的应用消息，解析 XML 提取标题
+                        root = ET.fromstring(content)
+                        title_elem = root.find('.//title')
+                        if title_elem is not None:
+                            content = title_elem.text
+                except Exception as e:
+                    logger.error(f"解析消息类型49出错: {e}")
+
+                sender_name = all_contacts.get(sender, sender)  # 获取发送者的名字，若找不到则使用原 ID
+                result.append(f"{timestamp},{sender_name},{content}")
+
+            result_str = "\n".join(result)  # 将结果拼接为最终字符串
+            return result_str
+
+        except Exception as e:
+            logger.error(f"获取消息出错: {e}")
+            return ""