279 lines
9.9 KiB
Python
279 lines
9.9 KiB
Python
from loguru import logger
|
||
import requests
|
||
import re
|
||
import os
|
||
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__()
|
||
|
||
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")
|
||
|
||
# 从TOML配置文件加载配置
|
||
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)
|
||
|
||
# 从TOML配置文件加载青龙面板配置
|
||
ql_host = self._config.get("JD_Token", {}).get("QL_HOST", "http://localhost:5700")
|
||
client_id = self._config.get("JD_Token", {}).get("CLIENT_ID", "")
|
||
client_secret = self._config.get("JD_Token", {}).get("CLIENT_SECRET", "")
|
||
|
||
# 初始化青龙面板连接
|
||
self.ql = QL(ql_host, client_id, client_secret)
|
||
|
||
self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}")
|
||
return 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.debug(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.JD_TOKEN) == PermissionStatus.DISABLED:
|
||
return False, "没有权限"
|
||
|
||
# 提取token和备注
|
||
token = match.group(1)
|
||
remark = match.group(2)
|
||
|
||
# 简单预检查token格式
|
||
if not token.startswith("pt_key=") or "pt_pin=" not in token:
|
||
await bot.send_text_message((roomid if roomid else sender),
|
||
f"❌ Token格式错误！正确格式应为：pt_key=xxx;pt_pin=xxx;",
|
||
sender)
|
||
return False, "Token格式错误"
|
||
|
||
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
|
||
env_id = None
|
||
for env in envs:
|
||
if env.get('name') == 'JD_COOKIE' and env.get('remarks') == remark:
|
||
existing_env = env
|
||
env_id = env.get('id')
|
||
break
|
||
|
||
result = False
|
||
self.LOG.debug(f"existing_env: {existing_env}")
|
||
if existing_env:
|
||
# 更新已存在的环境变量
|
||
existing_env['value'] = token
|
||
env_update: dict = {"id": env_id, "value": token, "remarks": remark,"name": "JD_COOKIE"}
|
||
result = self.ql.updateEnv(env_update)
|
||
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失败"
|