first commit

.gitignore

@@ -0,0 +1,5 @@
+napcat/
+.idea
+.venv
+data/
+logs/

config.toml

@@ -0,0 +1,30 @@
+#napcat配置
+bot_qq = 123456 #机器人q号
+root_qq = 1234567 # 管理员q号
+ws_uri = "loacalhost:3001" # ws 地址, 可自定义端口, 默认 3001
+webui_uri = "loacalhost:6099" # webui 地址, 可自定义端口, 默认 6099
+webui_token =  "napcat" # webui 令牌, 默认 napcat
+ws_token = "" # ws_uri 令牌, 默认留空
+ws_listen_ip = "localhost" # ws_uri 监听 ip, 默认 localhost 监听本机，监听全部则配置 0.0.0.0
+remote_mode = false # 是否远程模式, 即 NapCat 服务不在本机运行 ps：ncatbot官方已废弃该参数
+
+# 功能配置
+allowed_groups = "all" # 授权群聊，all为全部，eg：[123456789, 987654321]
+allowed_users = "all" # 授权用户，all为全部，eg：[123456789, 987654321]
+ai_service = "xyit" # ai平台 支持 “dify” “xyit"
+friend_auto = false # 好友自动同意
+group_auto = true
+group_welcome = false # 入群欢迎
+group_welcome_message = "!at 欢迎加入本群，使用@bot /help获取此bot帮助" # 入群消息 !at 为@加群用户
+group_leave = false # 退群提醒
+group_leave_message = "用户{userid}退群了" # 退群消息 ，{userid}为退群用户id
+
+# dify配置 ai_service选择dify时需配置
+dify_ip= "" # ip:端口
+dify_token = "" # token
+
+# xyit配置 ai_service选择xyit时需配置
+xyit_ip = "ai.openapi.xyit.net" # ip 此项一般不需用修改
+xyit_appID = "" # appID
+xyit_appKEY = "" # appKEY
+xyit_model = "maoniang" # 模型名称

control/group.py

@@ -0,0 +1,97 @@
+import logging
+from model.AiCat import AiCat
+from model.Clear import Clear
+import toml
+
+class group:
+   def __init__(self, msg):
+        self.user_id = msg.user_id
+        self.group_id = msg.group_id
+        self.message_id = msg.message_id
+        self.message_type = msg.message_type
+        self.raw_message = msg.raw_message
+        self.sender = msg.sender
+        self.message = msg.message
+        self.self_id = msg.self_id
+        self.time = msg.time
+
+   def main(self):
+       is_at = self.is_at()
+       if is_at is None:
+           return None
+       else:
+           permission = self.check_permission()
+           if permission is None:
+               return "服务器繁忙，请稍后再逝"
+           elif permission:
+               return self.menu(is_at)
+           else:
+               return "此bot未在该群启用"
+
+   def is_at(self):
+       for seg in self.message:
+           if seg['type'] == 'at' and seg['data'].get('qq') == str(self.self_id):
+               texts = [s['data']['text'].strip() for s in self.message if s['type'] == 'text']
+               full_text = ' '.join(texts).strip()
+               return full_text
+
+       return None
+
+   def check_permission(self):
+       try:
+           with open("./config.toml", "r", encoding="utf-8") as f:
+               config = toml.load(f)
+               allowed_groups = config.get("allowed_groups", [])
+       except Exception as e:
+           logging.error(str(e))
+           return None
+
+           # 检查当前群是否在允许列表中
+       if allowed_groups == "all":
+           return True
+       elif self.group_id in allowed_groups:
+           return True
+       else:
+           return False
+
+   def menu(self,command):
+        if command.startswith("/help"):
+            return " 直接输入聊天内容即可 \n /help -- 获取帮助 \n /clear [群号 / private:Q号]  (all 为全部，不填为本 群/用户)  \n /status -- 查看bot状态 "
+        elif command.startswith("/cat"):#  留此指令接口为了方便通过审核
+            # 排除 " /cat "" /cat"未传参情况
+            parts = command.split("/cat ", 1)
+            if len(parts) > 1:
+                chat_content = parts[1].strip()
+                if chat_content:
+                    cat = AiCat(chat_content, self.user_id,self.group_id)
+                    answer = AiCat.main(cat)
+                    return answer
+                else:
+                    return "你似乎没有提供想和我聊的内容喵~  \n 格式：/cat <提问内容>"
+            else:
+                return "你似乎没有提供想和我聊的内容喵~  \n 格式：/cat <提问内容>"
+        elif command.startswith("/clear"):
+            parts = command.split("/clear ", 1)
+            if len(parts) > 1:
+                group_id = parts[1].strip()
+                if group_id:
+                    clear = Clear(self.user_id,group_id)
+                    return clear.main()
+                else:
+                    clear = Clear(self.user_id, self.group_id)
+                    return  clear.main()
+            else:
+                clear = Clear(self.user_id, self.group_id)
+                return clear.main()
+        elif command.startswith("/status"):
+            return "---猫娘 QBOT---\n Q bot 运行正常 \n 版本: 2.0 pre \n © 融玩文化 | 无尽创意MCUNC"
+        elif command.startswith("/"):
+            return "指令不存在，输入/help查看帮助"
+        elif command == "":
+            return "你似乎没有提供想和我聊的内容喵~  \n 直接输入聊天内容即可"
+        elif command is None:
+            return "你似乎没有提供想和我聊的内容喵~  \n 直接输入聊天内容即可"
+        else:
+            cat = AiCat(command, self.user_id,self.group_id)
+            answer = AiCat.main(cat)
+            return answer

control/notice.py

@@ -0,0 +1,63 @@
+import toml
+import logging
+
+class notice:
+    def __init__(self,msg):
+        print(msg)
+        if msg["notice_type"] == "group_increase" or msg["notice_type"] == "group_decrease":
+            self.time = msg["time"]
+            self.self_id = msg["self_id"]
+            self.post_type = msg["post_type"]
+            self.notice_type = msg["notice_type"]
+            self.sub_type = msg["sub_type"]
+            self.group_id = msg["group_id"]
+            self.operator_id = msg["operator_id"]
+            self.user_id = msg["user_id"]
+        else:
+            pass
+
+    def main(self):
+        if self.notice_type == "group_increase":
+            return self.group_increase()
+        elif self.notice_type == "group_decrease":
+            return self.group_decrease()
+        else:
+            return None
+
+    def group_increase(self):
+        print(1)
+        try:
+            with open("./config.toml", "r", encoding="utf-8") as f:
+                config = toml.load(f)
+                group_welcome = config.get("group_welcome")
+                print(group_welcome)
+                group_welcome_message = config.get("group_welcome_message")
+                print(group_welcome_message)
+                if group_welcome:
+                    if "!at" in group_welcome_message:
+                        return group_welcome_message.replace("!at", f"[CQ:at,qq={self.user_id}]")
+                    else:
+                        return group_welcome_message
+                else:
+                    return None
+        except Exception as e:
+            logging.error(f"读取配置文件错误：{e}")
+            return None
+
+    def group_decrease(self):
+        try:
+            with open("./config.toml", "r", encoding="utf-8") as f:
+                config = toml.load(f)
+                group_leave = config.get("group_leave")
+                group_leave_message = config.get("group_leave_message")
+                print(group_leave_message)
+                if group_leave:
+                    if "{userid}" in group_leave_message:
+                        return group_leave_message.replace("{userid}", str(self.user_id))
+                    else:
+                        return group_leave_message
+                else:
+                    return None
+        except Exception as e:
+            logging.error(f"读取配置文件错误：{e}")
+            return None

control/private.py

@@ -0,0 +1,85 @@
+import logging
+from model.AiCat import AiCat
+from model.Clear import Clear
+import toml
+
+class private:
+   def __init__(self, msg):
+       self.user_id = msg.user_id
+       self.message_id = msg.message_id
+       self.message_type = msg.message_type
+       self.raw_message = msg.raw_message
+       self.sender = msg.sender
+       self.message = msg.message
+       self.self_id = msg.self_id
+       self.time = msg.time
+
+   def main(self):
+       texts = [seg['data']['text'].strip() for seg in self.message if seg['type'] == 'text']
+       full_text = ' '.join(texts).strip()
+       permission = self.check_permission()
+       if permission is None:
+           return "服务器繁忙，请稍后再逝"
+       elif permission:
+           return self.menu(full_text)
+       else:
+           return "此bot未在该群启用"
+
+   def check_permission(self):
+       try:
+           with open("./config.toml", "r", encoding="utf-8") as f:
+               config = toml.load(f)
+               allowed_users = config.get("allowed_users", [])
+       except Exception as e:
+           logging.error(str(e))
+           return None
+
+           # 检查当前群是否在允许列表中
+       if allowed_users == "all":
+           return True
+       elif self.user_id in allowed_users:
+           return True
+       else:
+           return False
+
+   def menu(self,command):
+        if command.startswith("/help"):
+            return " 直接输入聊天内容即可 \n /help -- 获取帮助 \n /clear [群号 / private:Q号]  (all 为全部，不填为本 群/用户)  \n /status -- 查看bot状态 "
+        elif command.startswith(" /cat"):#  留此指令接口为了方便通过审核
+            # 排除 " /cat "" /cat"未传参情况
+            parts = command.split(" /cat ", 1)
+            if len(parts) > 1:
+                chat_content = parts[1].strip()
+                if chat_content:
+                    cat = AiCat(chat_content,self.user_id , f"private:{self.user_id}")
+                    answer = AiCat.main(cat)
+                    return answer
+                else:
+                    return "你似乎没有提供想和我聊的内容喵~  \n 格式：/cat <提问内容>"
+            else:
+                return "你似乎没有提供想和我聊的内容喵~  \n 格式：/cat <提问内容>"
+        elif command.startswith("/clear"):
+            parts = command.split("/clear ", 1)
+            if len(parts) > 1:
+                group_id = parts[1].strip()
+                if group_id:
+                    clear = Clear(self.user_id,group_id)
+                    return clear.main()
+                else:
+                    clear = Clear(self.user_id, f"private:{self.user_id}")
+                    return  clear.main()
+            else:
+                clear = Clear(self.user_id, f"private:{self.user_id}")
+                return clear.main()
+        elif command.startswith(" /status"):
+            return "---猫娘 QBOT---\n Q bot 运行正常 \n 版本: 2.0 pre \n © 融玩文化 | 无尽创意MCUNC"
+        elif command.startswith(" /"):
+            return "指令不存在，输入/help查看帮助"
+        elif command == "":
+            return "你似乎没有提供想和我聊的内容喵~  \n 直接输入聊天内容即可"
+        elif command is None:
+            return "你似乎没有提供想和我聊的内容喵~  \n 直接输入聊天内容即可"
+        else:
+            cat = AiCat(command, self.user_id,f"private:{self.user_id}")
+            answer = AiCat.main(cat)
+            return answer

control/request.py

@@ -0,0 +1,42 @@
+import logging
+import toml
+
+class request:
+    def __init__(self,msg):
+        self.time = msg.time
+        self.self_id = msg.self_id
+        self.request_type = msg.request_type
+        self.sub_type = msg.sub_type
+        self.group_id = msg.group_id
+        self.user_id = msg.user_id
+        self.comment = msg.comment
+        self.flag = msg.flag
+
+    def main(self):
+        if self.request_type == "friend":
+            friend_auto = self.get_info()
+            if friend_auto:
+                return True
+            else:
+                return False
+
+
+    def get_info(self):
+        try:
+            with open("./config.toml", "r", encoding="utf-8") as f:
+                config = toml.load(f)
+                friend_auto = config.get("friend_auto")
+            return friend_auto
+        except Exception as e:
+            logging.error(f"读取配置文件错误：{e}")
+            return False
+
+    def get_allow_group(self):
+        try:
+            with open("./config.toml", "r", encoding="utf-8") as f:
+                config = toml.load(f)
+                allow_group = config.get("allowed_groups")
+            return allow_group
+        except Exception as e:
+            logging.error(f"读取配置文件错误：{e}")
+            return []

main.py

@@ -0,0 +1,83 @@
+from ncatbot.core.notice import NoticeMessage
+from ncatbot.core import BotClient, Request
+import logging
+import asyncio
+from control.group import group
+from control.private import private
+from control.request import request
+from control.notice import notice
+import toml
+
+
+with open("./config.toml", "r", encoding="utf-8") as f:
+    config = toml.load(f)
+    bt_uin = config.get("bot_qq")
+    root = config.get("root_qq")
+    ws_uri = config.get("ws_uri")
+    web_uri = config.get("web_uri")
+    webui_token = config.get("webui_token")
+    ws_token = config.get("ws_token")
+    ws_listen_ip = config.get("ws_listen_ip")
+    remote_mode = config.get("remote_mode")
+
+bot = BotClient()
+api = bot.run_blocking(bt_uin=bt_uin, root=root, ws_uri=ws_uri, web_uri=web_uri, webui_token=webui_token, ws_token=ws_token, ws_listen_ip=ws_listen_ip, remote_mode=remote_mode)
+
+@bot.group_event()
+async def on_group_message(msg):
+    logging.info(f"收到消息：{msg.raw_message}，来自{msg.group_id}群聊{msg.user_id}用户")
+    if msg.user_id == 2854196310: # qq管家，防止刷屏。bot大战请看
+        pass
+    else:
+        ctrl = group(msg)
+        return_message = ctrl.main()
+        if return_message is None:
+            return
+        else:
+            logging.info(f"返回消息：{return_message}")
+            await  bot.api.post_group_msg(group_id=msg.group_id, text=return_message, reply=msg.message_id)
+
+@bot.private_event()
+async def on_private_message(msg):
+    logging.info(f"收到消息：{msg.raw_message}，来自{msg.user_id}用户")
+    if msg.user_id == 2854196310: # qq管家，防止刷屏。
+        pass
+    else:
+        ctrl = private(msg)
+        return_message = ctrl.main()
+        if return_message is None:
+            return
+        else:
+            logging.info(f"返回消息：{return_message}")
+            await  bot.api.post_private_msg(user_id=msg.user_id, text=return_message, reply=msg.message_id)
+
+
+@bot.request_event()
+async def on_request_event(msg: Request):
+    logging.info(f"收到request事件：{msg.request_type}，来自{msg.group_id}群聊，{msg.user_id}用户，验证消息：{msg.comment}")
+    ctrl = request(msg)
+    accept_friend_application = ctrl.main()
+    if accept_friend_application is True:
+        await msg.reply(True, comment="请求已通过")
+        logging.info("请求已通过")
+    else:
+        await msg.reply(False, comment="请求被拒绝")
+        logging.info("请求被拒绝")
+
+
+
+@bot.notice_event()
+async  def on_notice_event(msg: NoticeMessage):
+    logging.info(f"收到notice事件：{msg['notice_type']}，来自{msg['user_id']}用户")
+    ctrl = notice(msg)
+    return_message = ctrl.main()
+    if return_message is None:
+        return
+    else:
+        logging.info(f"返回消息：{return_message}")
+        await  bot.api.post_group_msg(group_id=msg["group_id"], text=return_message)
+
+
+
+
+asyncio.get_event_loop().run_forever()

model/AiCat.py

@@ -0,0 +1,144 @@
+import requests
+import json
+import logging
+import toml
+import sqlite3
+
+class AiCat:
+    def __init__(self,message,qid,group_openid):
+        self.message = message
+        self.qid = qid
+        self.group_openid = group_openid
+        self.query = f"<qid>{self.qid}</qid>{self.message}"
+
+
+
+    def main(self):
+        with open('./config.toml', 'r', encoding='utf-8') as f:
+            config = toml.load(f)
+
+        # 获取所需字段
+        ai_service = config.get("ai_service")
+        # dify
+        dify_ip = config.get("dify_ip")
+        dify_token = config.get("dify_token")
+        # xyit
+        xyit_ip = config.get("xyit_ip")
+        xyit_appID = config.get("xyit_appID")
+        xyit_appKEY = config.get("xyit_appKEY")
+        xyit_model = config.get("xyit_model")
+
+        self.init_db()
+
+        uuid = self.get_uuid()
+
+        if uuid == "":
+            logging.info("未找到 UUID")
+        else:
+            logging.info(f"找到 UUID: {uuid}")
+
+        if ai_service == "dify":
+            # API URL
+            url = f"https://{dify_ip}/v1/chat-messages"  # 替换为实际的 API 地址
+
+            # 请求头
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": dify_token  # 替换为你的 API 密钥
+            }
+
+            # 请求体
+            payload = {
+                "query": self.query,  # 用户输入/提问内容
+                "inputs": {},  # App 定义的变量值（默认为空）
+                "response_mode": "blocking",  # 流式模式或阻塞模式
+                "user": "QBotAPI",  # 用户标识，需保证唯一性
+                "conversation_id": uuid,  # （选填）会话 ID，继续对话时需要传入
+                "files": [],  # 文件列表（选填），适用于文件结合文本理解
+                "auto_generate_name": True  # （选填）自动生成标题，默认为 True
+            }
+
+            logging.info("请求平台：dify")
+
+        elif ai_service == "xyit":
+            url = f"http://{xyit_ip}/{xyit_model}/"  # 替换为实际的 API 地址
+
+            # 请求头
+            headers = {
+                "Content-Type": "application/json",
+                "appID": xyit_appID ,  # 替换为你的 API 密钥
+                "appKEY": xyit_appKEY
+            }
+
+            # 请求体
+            payload = {
+                "query": self.query,  # 用户输入/提问内容
+                "conversation_id": uuid,  # （选填）会话 ID，继续对话时需要传入
+            }
+
+            logging.info("请求平台：xyit")
+
+        else:
+            logging.error("未配置ai平台")
+            return "服务器繁忙，请稍后再逝"
+
+        # 发送 POST 请求
+        try:
+            response = requests.post(url, headers=headers, data=json.dumps(payload))
+
+            # 检查响应状态码
+            if response.status_code == 200:
+                logging.info("请求成功！返回结果：")
+                response_data = response.json()
+                print(response_data) # test
+                # 存储uuid
+                if uuid == "":
+                    try:
+                        with sqlite3.connect("uuid.db") as conn:
+                            cursor = conn.cursor()
+                            cursor.execute("INSERT INTO groups (group_openid, uuid) VALUES (?, ?)",
+                                           (self.group_openid, response_data.get("conversation_id")))
+                    except sqlite3.Error as e:
+                        logging.error(f"数据库插入错误: {e}")
+
+                # 提取 answer 值
+                answer = response_data.get("answer","服务器繁忙，请稍后再逝")
+                return answer
+            else:
+                logging.info(f"请求失败！状态码: {response.status_code}")
+                logging.info(f"错误信息: {response.text}")  # 打印错误信息
+                return "服务器繁忙，请稍后再逝"
+
+        except Exception as e:
+            logging.info(f"请求过程中出现异常: {e}")
+            return "服务器繁忙，请稍后再逝"
+
+
+    def get_uuid(self):
+        try:
+            with sqlite3.connect("uuid.db") as conn:
+                cursor = conn.cursor()
+                cursor.execute("SELECT uuid FROM groups WHERE group_openid = ?", (self.group_openid,))
+                result = cursor.fetchone()
+                return result[0] if result else ""
+        except sqlite3.Error as e:
+            logging.error(f"数据库查询错误: {e}")
+            return ""
+
+    def init_db(self):
+        try:
+            with sqlite3.connect("uuid.db") as conn:
+                cursor = conn.cursor()
+                cursor.execute('''
+                    CREATE TABLE IF NOT EXISTS groups (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        group_openid TEXT UNIQUE NOT NULL,
+                        uuid TEXT NOT NULL
+                    )
+                ''')
+                conn.commit()
+        except sqlite3.Error as e:
+            logging.error(f"数据库初始化失败: {e}")
+
+
+

model/Clear.py

@@ -0,0 +1,56 @@
+import logging
+import toml
+import sqlite3
+
+
+class Clear:
+    def __init__(self, user_id,group_id):
+        self.user_id = user_id
+        self.group_id = group_id
+
+    def main(self):
+        if self.is_root():
+            if self.group_id == "all":
+                try:
+                    with sqlite3.connect("uuid.db") as conn:
+                        cursor = conn.cursor()
+                        # 清空表中所有数据
+                        cursor.execute("DELETE FROM groups;")  # 假设表名为 uuid_table，请根据实际表名修改
+                        conn.commit()
+                        logging.info("✅ 数据库表已成功清空")
+                        return "✅ 已清空所有群组数据"
+                except sqlite3.Error as e:
+                    if 'conn' in locals():
+                        conn.close()
+                    logging.error(f"❌ 数据库操作失败: {e}")
+                    return "❌ 数据库操作失败"
+            else:
+                try:
+                    with sqlite3.connect("uuid.db") as conn:
+                        cursor = conn.cursor()
+                        # 删除指定 group_id 对应的数据行
+                        cursor.execute("DELETE FROM groups WHERE group_openid = ?", (self.group_id,))
+                        conn.commit()
+                        if cursor.rowcount > 0:
+                            logging.info(f"✅ 已成功删除 group_id = {self.group_id} 的数据")
+                            return f"✅ 已成功删除 group_id = {self.group_id} 的数据"
+                        else:
+                            logging.info(f"⚠️ 没有找到 group_id = {self.group_id} 的数据")
+                            return f"⚠️ 没有找到 group_id = {self.group_id} 的数据"
+                except sqlite3.Error as e:
+                    if 'conn' in locals():
+                        conn.close()
+                    logging.error(f"❌ 数据库操作失败: {e}")
+                    return "❌ 数据库操作失败"
+
+        else:
+            return "你不是管理员哦喵~"
+
+    def is_root(self):
+        with open("./config.toml", "r", encoding="utf-8") as f:
+            config = toml.load(f)
+            root = config.get("root_qq")
+        if self.user_id == root:
+            return True
+        else:
+            return False