|
|
import logging # 导入日志模块，用于记录程序运行过程中的日志信息
|
|
|
import os # 导入os模块，用于与操作系统交互，如获取环境变量、执行系统命令等
|
|
|
|
|
|
import openai # 导入openai模块，用于调用OpenAI的API服务
|
|
|
|
|
|
from servermanager.models import commands # 从servermanager应用的models模块中导入commands模型，用于操作命令相关的数据
|
|
|
|
|
|
logger = logging.getLogger(__name__) # 创建日志记录器，名称为当前模块名，用于记录该模块的日志
|
|
|
|
|
|
openai.api_key = os.environ.get('OPENAI_API_KEY') # 从环境变量中获取OpenAI的API密钥，并设置为openai模块的API密钥
|
|
|
if os.environ.get('HTTP_PROXY'): # 检查环境变量中是否设置了HTTP代理，如果有则为openai模块设置代理
|
|
|
openai.proxy = os.environ.get('HTTP_PROXY')
|
|
|
|
|
|
|
|
|
class ChatGPT:
|
|
|
"""
|
|
|
ChatGPT类，用于与OpenAI的GPT模型进行交互，实现聊天功能
|
|
|
"""
|
|
|
@staticmethod
|
|
|
def chat(prompt):
|
|
|
"""
|
|
|
静态方法，发送提示信息给GPT模型并获取回复
|
|
|
|
|
|
:param prompt: 用户输入的提示信息（字符串）
|
|
|
:return: GPT模型的回复内容（字符串），若出错则返回"服务器出错了"
|
|
|
"""
|
|
|
try:
|
|
|
# 调用OpenAI的ChatCompletion接口，使用gpt-3.5-turbo模型
|
|
|
# messages参数为消息列表，包含用户角色和对应的内容
|
|
|
completion = openai.ChatCompletion.create(model="gpt-3.5-turbo",
|
|
|
messages=[{"role": "user", "content": prompt}])
|
|
|
# 从返回结果中提取第一个选择的消息内容并返回
|
|
|
return completion.choices[0].message.content
|
|
|
except Exception as e:
|
|
|
# 捕获异常并记录错误日志
|
|
|
logger.error(e)
|
|
|
# 返回错误提示信息
|
|
|
return "服务器出错了"
|
|
|
|
|
|
|
|
|
class CommandHandler:
|
|
|
"""
|
|
|
命令处理器类，用于处理和执行系统命令，以及提供命令帮助信息
|
|
|
"""
|
|
|
def __init__(self):
|
|
|
"""
|
|
|
初始化方法，加载所有的命令数据
|
|
|
从commands模型中查询所有命令记录并存储在实例变量self.commands中
|
|
|
"""
|
|
|
self.commands = commands.objects.all()
|
|
|
|
|
|
def run(self, title):
|
|
|
"""
|
|
|
运行命令
|
|
|
:param title: 命令
|
|
|
:return: 返回命令执行结果
|
|
|
"""
|
|
|
# 使用filter函数筛选出标题（不区分大小写）与输入title匹配的命令
|
|
|
# 将筛选结果转换为列表
|
|
|
cmd = list(
|
|
|
filter(
|
|
|
lambda x: x.title.upper() == title.upper(),
|
|
|
self.commands))
|
|
|
# 如果找到匹配的命令
|
|
|
if cmd:
|
|
|
# 调用私有方法执行命令，传入命令的具体内容（cmd[0].command）
|
|
|
return self.__run_command__(cmd[0].command)
|
|
|
else:
|
|
|
# 未找到命令时，返回提示信息
|
|
|
return "未找到相关命令，请输入hepme获得帮助。"
|
|
|
|
|
|
def __run_command__(self, cmd):
|
|
|
"""
|
|
|
私有方法，用于执行具体的系统命令
|
|
|
|
|
|
:param cmd: 要执行的系统命令（字符串）
|
|
|
:return: 命令执行的输出结果（字符串）；若执行出错，返回错误提示
|
|
|
"""
|
|
|
try:
|
|
|
# 使用os.popen执行命令，并读取命令的输出结果
|
|
|
res = os.popen(cmd).read()
|
|
|
return res
|
|
|
except BaseException:
|
|
|
# 捕获所有基本异常，返回命令执行出错的提示
|
|
|
return '命令执行出错!'
|
|
|
|
|
|
def get_help(self):
|
|
|
"""
|
|
|
获取所有命令的帮助信息
|
|
|
|
|
|
:return: 包含所有命令标题和描述的字符串，每条命令占一行
|
|
|
"""
|
|
|
rsp = ''
|
|
|
# 遍历所有命令，拼接命令标题和描述信息
|
|
|
for cmd in self.commands:
|
|
|
rsp += '{c}:{d}\n'.format(c=cmd.title, d=cmd.describe)
|
|
|
return rsp
|
|
|
|
|
|
# 当该模块作为主程序运行时执行以下代码
|
|
|
if __name__ == '__main__':
|
|
|
chatbot = ChatGPT()
|
|
|
prompt = "写一篇1000字关于AI的论文"
|
|
|
print(chatbot.chat(prompt))
|
