codfrm/MoviePilot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add chatgpt plugin

Browse Source
This commit is contained in:
jxxghp 2023-08-06 18:57:36 +08:00
parent 6bf757dc68
commit 6a4b5e6549
4 changed files with 345 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
app/plugins/chatgpt/__init__.py Normal file
View File

@ -0,0 +1,140 @@
from typing import Any, List, Dict, Tuple
from app.core.event import eventmanager
from app.plugins import _PluginBase
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
# 私有属性
_enabled = False
_openai_url = None
_openai_key = None
def init_plugin(self, config: dict = None):
if config:
self._enabled = config.get("enabled")
self._openai_url = config.get("openai_url")
self._openai_key = config.get("openai_key")
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': '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,
"openai_url": "https://api.openai.com",
"openai_key": ""
}
def get_page(self) -> List[dict]:
pass
@eventmanager.register(EventType.UserMessage)
def talk(self, event):
"""
监听用户消息，获取ChatGPT回复
"""
pass
def stop_service(self):
"""
退出插件
"""
pass

200
app/plugins/chatgpt/openai.py Normal file
View File

@ -0,0 +1,200 @@
import json
import time
from typing import List, Union
import openai
from cacheout import Cache
OpenAISessionCache = Cache(maxsize=100, ttl=3600, timer=time.time, default=None)
class OpenAi:
_api_key: str = None
_api_url: str = None
def __init__(self, api_key: str = None, api_url: str = None):
self._api_key = api_key
self._api_url = api_url
def get_state(self) -> bool:
return True if self._api_key else False
@staticmethod
def __save_session(session_id: str, message: str):
"""
保存会话
:param session_id: 会话ID
:param message: 消息
:return:
"""
seasion = OpenAISessionCache.get(session_id)
if seasion:
seasion.append({
"role": "assistant",
"content": message
})
OpenAISessionCache.set(session_id, seasion)
@staticmethod
def __get_session(session_id: str, message: str) -> List[dict]:
"""
获取会话
:param session_id: 会话ID
:return: 会话上下文
"""
seasion = OpenAISessionCache.get(session_id)
if seasion:
seasion.append({
"role": "user",
"content": message
})
else:
seasion = [
{
"role": "system",
"content": "请在接下来的对话中请使用中文回复，并且内容尽可能详细。"
},
{
"role": "user",
"content": message
}]
OpenAISessionCache.set(session_id, seasion)
return seasion
@staticmethod
def __get_model(message: Union[str, List[dict]],
prompt: str = None,
user: str = "MoviePilot",
**kwargs):
"""
获取模型
"""
if not isinstance(message, list):
if prompt:
message = [
{
"role": "system",
"content": prompt
},
{
"role": "user",
"content": message
}
]
else:
message = [
{
"role": "user",
"content": message
}
]
return openai.ChatCompletion.create(
model="gpt-3.5-turbo",
user=user,
messages=message,
**kwargs
)
@staticmethod
def __clear_session(session_id: str):
"""
清除会话
:param session_id: 会话ID
:return:
"""
if OpenAISessionCache.get(session_id):
OpenAISessionCache.delete(session_id)
def get_media_name(self, filename: str):
"""
从文件名中提取媒体名称等要素
:param filename: 文件名
:return: Json
"""
if not self.get_state():
return None
result = ""
try:
_filename_prompt = "I will give you a movie/tvshow file name.You need to return a Json." \
"\nPay attention to the correct identification of the film name." \
"\n{\"title\":string,\"version\":string,\"part\":string,\"year\":string,\"resolution\":string,\"season\":number|null,\"episode\":number|null}"
completion = self.__get_model(prompt=_filename_prompt, message=filename)
result = completion.choices[0].message.content
return json.loads(result)
except Exception as e:
print(f"{str(e)}：{result}")
return {}
def get_response(self, text: str, userid: str):
"""
聊天对话，获取答案
:param text: 输入文本
:param userid: 用户ID
:return:
"""
if not self.get_state():
return ""
try:
if not userid:
return "用户信息错误"
else:
userid = str(userid)
if text == "#清除":
self.__clear_session(userid)
return "会话已清除"
# 获取历史上下文
messages = self.__get_session(userid, text)
completion = self.__get_model(message=messages, user=userid)
result = completion.choices[0].message.content
if result:
self.__save_session(userid, text)
return result
except openai.error.RateLimitError as e:
return f"请求被ChatGPT拒绝了，{str(e)}"
except openai.error.APIConnectionError as e:
return f"ChatGPT网络连接失败：{str(e)}"
except openai.error.Timeout as e:
return f"没有接收到ChatGPT的返回消息：{str(e)}"
except Exception as e:
return f"请求ChatGPT出现错误：{str(e)}"
def translate_to_zh(self, text: str):
"""
翻译为中文
:param text: 输入文本
"""
if not self.get_state():
return False, None
system_prompt = "You are a translation engine that can only translate text and cannot interpret it."
user_prompt = f"translate to zh-CN:\n\n{text}"
result = ""
try:
completion = self.__get_model(prompt=system_prompt,
message=user_prompt,
temperature=0,
top_p=1,
frequency_penalty=0,
presence_penalty=0)
result = completion.choices[0].message.content.strip()
return True, result
except Exception as e:
print(f"{str(e)}：{result}")
return False, str(e)
def get_question_answer(self, question: str):
"""
从给定问题和选项中获取正确答案
:param question: 问题及选项
:return: Json
"""
if not self.get_state():
return None
result = ""
try:
_question_prompt = "下面我们来玩一个游戏，你是老师，我是学生，你需要回答我的问题，我会给你一个题目和几个选项，你的回复必须是给定选项中正确答案对应的序号，请直接回复数字"
completion = self.__get_model(prompt=_question_prompt, message=question)
result = completion.choices[0].message.content
return result
except Exception as e:
print(f"{str(e)}：{result}")
return {}

2
app/schemas/types.py
View File

@ -34,6 +34,8 @@ class EventType(Enum):
HistoryDeleted = "history.deleted" HistoryDeleted = "history.deleted"
# 删除媒体库文件 # 删除媒体库文件
MediaDeleted = "media.deleted" MediaDeleted = "media.deleted"
# 用户外来消息
UserMessage = "user.message"
# 系统配置Key字典 # 系统配置Key字典

2
requirements.txt
View File

@ -46,3 +46,5 @@ psutil~=5.9.4
python_hosts~=1.0.3 python_hosts~=1.0.3
watchdog~=3.0.0 watchdog~=3.0.0
tailer~=0.4.1 tailer~=0.4.1
openai~=0.27.2
cacheout~=0.14.1