from typing import Any, List, Dict, Tuple from app.core.config import settings from app.core.event import eventmanager from app.plugins import _PluginBase from app.plugins.chatgpt.openai import OpenAi from app.schemas.types import EventType class ChatGPT(_PluginBase): # 插件名称 plugin_name = "ChatGPT" # 插件描述 plugin_desc = "消息交互支持与ChatGPT对话。" # 插件图标 plugin_icon = "chatgpt.png" # 主题色 plugin_color = "#74AA9C" # 插件版本 plugin_version = "1.0" # 插件作者 plugin_author = "jxxghp" # 作者主页 author_url = "https://github.com/jxxghp" # 插件配置项ID前缀 plugin_config_prefix = "chatgpt_" # 加载顺序 plugin_order = 15 # 可使用的用户级别 auth_level = 1 # 私有属性 openai = None _enabled = False _proxy = False _openai_url = None _openai_key = None def init_plugin(self, config: dict = None): if config: self._enabled = config.get("enabled") self._proxy = config.get("proxy") self._openai_url = config.get("openai_url") self._openai_key = config.get("openai_key") self.openai = OpenAi(api_key=self._openai_key, api_url=self._openai_url, proxy=settings.PROXY if self._proxy else None) def get_state(self) -> bool: return self._enabled @staticmethod def get_command() -> List[Dict[str, Any]]: pass def get_api(self) -> List[Dict[str, Any]]: pass def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: """ 拼装插件配置页面,需要返回两块数据:1、页面配置;2、数据结构 """ return [ { 'component': 'VForm', 'content': [ { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'enabled', 'label': '启用插件', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'proxy', 'label': '使用代理', } } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'openai_url', 'label': 'OpenAI API Url', 'placeholder': 'https://api.openai.com', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'openai_key', 'label': 'sk-xxx' } } ] } ] }, ] } ], { "enabled": False, "proxy": False, "openai_url": "https://api.openai.com", "openai_key": "" } def get_page(self) -> List[dict]: pass @eventmanager.register(EventType.UserMessage) def talk(self, event): """ 监听用户消息,获取ChatGPT回复 """ if not self.openai: return text = event.event_data.get("text") userid = event.event_data.get("userid") channel = event.event_data.get("channel") if not text: return response = self.openai.get_response(text=text, userid=userid) if response: self.post_message(channel=channel, title=response, userid=userid) def stop_service(self): """ 退出插件 """ pass