from loguru import logger import requests import re import os import yaml from json import dumps as jsonDumps from typing import Dict, Any, List, Optional, Tuple from plugin_common.message_plugin_interface import MessagePluginInterface from plugin_common.plugin_interface import PluginStatus from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager from utils.decorator.points_decorator import plugin_points_cost from wechat_ipad import WechatAPIClient class QL: def __init__(self, address: str, id: str, secret: str) -> None: """ 初始化 """ self.address = address self.id = id self.secret = secret self.valid = True self.login() def log(self, content: str) -> None: """ 日志 """ print(content) def login(self) -> None: """ 登录 """ url = f"{self.address}/open/auth/token?client_id={self.id}&client_secret={self.secret}" try: rjson = requests.get(url).json() if (rjson['code'] == 200): self.auth = f"{rjson['data']['token_type']} {rjson['data']['token']}" else: self.log(f"登录失败：{rjson['message']}") except Exception as e: self.valid = False self.log(f"登录失败：{str(e)}") def getEnvs(self) -> list: """ 获取环境变量 """ url = f"{self.address}/open/envs?searchValue=" headers = {"Authorization": self.auth} try: rjson = requests.get(url, headers=headers).json() if (rjson['code'] == 200): return rjson['data'] else: self.log(f"获取环境变量失败：{rjson['message']}") except Exception as e: self.log(f"获取环境变量失败：{str(e)}") def deleteEnvs(self, ids: list) -> bool: """ 删除环境变量 """ url = f"{self.address}/open/envs" headers = {"Authorization": self.auth, "content-type": "application/json"} try: rjson = requests.delete(url, headers=headers, data=jsonDumps(ids)).json() if (rjson['code'] == 200): self.log(f"删除环境变量成功：{len(ids)}") return True else: self.log(f"删除环境变量失败：{rjson['message']}") return False except Exception as e: self.log(f"删除环境变量失败：{str(e)}") return False def addEnvs(self, envs: list) -> bool: """ 新建环境变量 """ url = f"{self.address}/open/envs" headers = {"Authorization": self.auth, "content-type": "application/json"} try: rjson = requests.post(url, headers=headers, data=jsonDumps(envs)).json() if (rjson['code'] == 200): self.log(f"新建环境变量成功：{len(envs)}") return True else: self.log(f"新建环境变量失败：{rjson['message']}") return False except Exception as e: self.log(f"新建环境变量失败：{str(e)}") return False def updateEnv(self, env: dict) -> bool: """ 更新环境变量 """ url = f"{self.address}/open/envs" headers = {"Authorization": self.auth, "content-type": "application/json"} try: rjson = requests.put(url, headers=headers, data=jsonDumps(env)).json() if (rjson['code'] == 200): self.log(f"更新环境变量成功") return True else: self.log(f"更新环境变量失败：{rjson['message']}") return False except Exception as e: self.log(f"更新环境变量失败：{str(e)}") return False class JDTokenPlugin(MessagePluginInterface): """京东签到Token设置插件""" @property def name(self) -> str: return "京东签到Token设置" @property def version(self) -> str: return "1.0.0" @property def description(self) -> str: return "提供京东签到Token设置功能，支持添加和更新Token" @property def author(self) -> str: return "Trae AI" @property def command_prefix(self) -> Optional[str]: return "" # 不需要前缀，直接匹配命令 @property def commands(self) -> List[str]: return self._commands def __init__(self): super().__init__() self.plugin_dir = os.path.dirname(os.path.abspath(__file__)) self.config_path = os.path.join(self.plugin_dir, "config.yaml") def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" self.LOG = logger self.LOG.info(f"正在初始化 {self.name} 插件...") # 保存上下文对象 self.event_system = context.get("event_system") self.message_util = context.get("message_util") self._commands = self._config.get("JD_Token", {}).get("command", ["设置京东"]) self.command_format = self._config.get("JD_Token", {}).get("command-format", "设置京东 token内容 备注名称") self.enable = self._config.get("JD_Token", {}).get("enable", True) # 加载青龙面板配置 self.load_config() self.ql = QL(self.ql_config.get("QL_HOST"), self.ql_config.get("CLIENT_ID"), self.ql_config.get("CLIENT_SECRET")) self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}") return True def load_config(self): """加载配置""" if os.path.exists(self.config_path): with open(self.config_path, 'r', encoding='utf-8') as f: self.ql_config = yaml.safe_load(f) else: # 默认配置 self.ql_config = { "QL_HOST": "http://localhost:5700", "CLIENT_ID": "", "CLIENT_SECRET": "" } # 保存默认配置 with open(self.config_path, 'w', encoding='utf-8') as f: yaml.dump(self.ql_config, f, default_flow_style=False, allow_unicode=True) def start(self) -> bool: """启动插件""" self.LOG.info(f"[{self.name}] 插件已启动") self.status = PluginStatus.RUNNING return True def stop(self) -> bool: """停止插件""" self.LOG.info(f"[{self.name}] 插件已停止") self.status = PluginStatus.STOPPED return True def can_process(self, message: Dict[str, Any]) -> bool: """检查是否可以处理该消息""" if not self.enable: return False content = str(message.get("content", "")).strip() command = content.split(" ")[0] return command in self._commands @plugin_stats_decorator(plugin_name="京东签到Token设置") async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.info(f"插件执行： {self.name}：{content}") command = content.split(" ")[0] sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") # 检查命令格式 pattern = r'^设置京东\s+([^\s]+)\s+(.+)$' match = re.match(pattern, content) if not match: await bot.send_text_message((roomid if roomid else sender), f"❌命令格式错误！\n{self.command_format}" , sender) return False, "命令格式错误" # 检查权限 if roomid and gbm.get_group_permission(roomid, Feature.UTILITY) == PermissionStatus.DISABLED: return False, "没有权限" # 提取token和备注 token = match.group(1) remark = match.group(2) try: # 设置京东Token result = self.set_jd_token(token, remark) await bot.send_text_message((roomid if roomid else sender), result, sender) return True, "处理成功" except Exception as e: self.LOG.error(f"处理京东Token设置请求出错: {e}") await bot.send_text_message((roomid if roomid else sender), f"❌处理出错: {str(e)}", sender) return False, f"处理出错: {e}" def set_jd_token(self, token: str, remark: str) -> str: """设置京东Token""" if not self.ql.valid: return f"❌ 青龙面板连接失败，请检查配置" # 检查是否已存在相同备注的环境变量 envs = self.ql.getEnvs() if not envs: return f"❌ 获取环境变量失败" # 查找是否有相同备注的JD_COOKIE existing_env = None for env in envs: if env.get('name') == 'JD_COOKIE' and env.get('remarks') == remark: existing_env = env break result = False if existing_env: # 更新已存在的环境变量 existing_env['value'] = token result = self.ql.updateEnv(existing_env) if result: return f"✅ 已成功更新京东账号 [{remark}] 的Token" else: return f"❌ 更新京东账号 [{remark}] 的Token失败" else: # 添加新的环境变量 new_env = [{"name": "JD_COOKIE", "value": token, "remarks": remark}] result = self.ql.addEnvs(new_env) if result: return f"✅ 已成功添加京东账号 [{remark}] 的Token" else: return f"❌ 添加京东账号 [{remark}] 的Token失败"