-
Notifications
You must be signed in to change notification settings - Fork 17
/
main.py
198 lines (150 loc) · 7.49 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
import traceback
from pkg.plugin.models import *
from pkg.plugin.host import EventContext, PluginHost, emit
import time
import os
from revChatGPT.V1 import Chatbot
import plugins.revLibs.pkg.process.revss as revss
import plugins.revLibs.pkg.process.procmsg as procmsg
import plugins.revLibs.pkg.process.proccmd as proccmd
from EdgeGPT.EdgeGPT import ConversationStyle
def check_config():
this_file = __file__
template_file = os.path.join(os.path.dirname(this_file), "revcfg-template.py")
# 检查revlib.py是否存在
if not os.path.exists("revcfg.py"):
# 不存在则使用本模块同目录的revcfg-template.py复制创建
with open(template_file, "r", encoding="utf-8") as f:
template = f.read()
with open("revcfg.py", "w", encoding="utf-8") as f:
f.write(template)
return False
return True
# 注册插件
@register(name="revLibs", description="接入acheong08/ChatGPT、Claude、Bard等逆向库", version="0.9.1", author="RockChinQ")
class RevLibsPlugin(Plugin):
chatbot: Chatbot = None
# 插件加载时触发
def __init__(self, plugin_host: PluginHost):
# 执行依赖库更新
# try:
# import plugins.revLibs.pkg.utils as utils
# # utils.upgrade_revlibs()
# except:
# traceback.print_exc()
# logging.warn("[rev] 依赖库更新失败，不能确保逆向库正常运行")
if not check_config():
logging.error("[rev] 已生成配置文件(revcfg.py)，请按照其中注释填写配置文件后重启程序")
# plugin_host.notify_admin("[rev] 已生成配置文件(revcfg.py)，请按照其中注释填写配置文件后重启程序")
return
import revcfg
if not hasattr(revcfg, "new_bing_style"):
setattr(revcfg, "new_bing_style", ConversationStyle.balanced)
try:
from plugins.revLibs.pkg.process.impls import v1impl, edgegpt, hugchat, claude, bard, gpt4free
reverse_lib_mapping = {
"acheong08/ChatGPT.V1": v1impl.RevChatGPTV1,
"acheong08/EdgeGPT": edgegpt.EdgeGPTImpl,
"Soulter/hugging-chat-api": hugchat.HugChatImpl,
"KoushikNavuluri/Claude-API": claude.ClaudeImpl,
"dsdanielpark/Bard-API": bard.BardImpl,
"xtekky/gpt4free": gpt4free.GPT4FreeImpl,
}
import plugins.revLibs.pkg.process.revss as revss
if hasattr(revcfg, "reverse_lib") and revcfg.reverse_lib in reverse_lib_mapping:
revss.__rev_interface_impl_class__ = reverse_lib_mapping[revcfg.reverse_lib]
logging.info("[rev] 已加载逆向库{}, 使用接口实现类: {}".format(revcfg.reverse_lib, str(reverse_lib_mapping[revcfg.reverse_lib])))
else:
logging.error("[rev] 未知的逆向库: " + revcfg.reverse_lib + ", 请检查配置文件是否填写正确或尝试更新逆向库插件")
time.sleep(5)
return
except:
# 输出完整的错误信息
# plugin_host.notify_admin("[rev] 逆向库初始化失败，请检查配置文件(revcfg.py)是否正确")
logging.error("[rev] 逆向库初始化失败，请检查配置文件(revcfg.py)是否正确")
logging.error("[rev] " + traceback.format_exc())
return
import config
revcfg.process_message_timeout = config.process_message_timeout
config.process_message_timeout = 10*60
logging.info("[rev] 已将主程序消息处理超时时间设置为10分钟")
@on(PersonNormalMessageReceived)
@on(GroupNormalMessageReceived)
def normal_message_received(inst, event: EventContext, **kwargs):
event.prevent_default()
event.prevent_postorder()
reply = []
try:
prefix = revcfg.reply_prefix
reply_message = ""
reply_message = procmsg.process_message(session_name=kwargs['launcher_type']+"_"+str(kwargs['launcher_id']),
prompt=kwargs['text_message'], **kwargs)
logging.debug("[rev] " + reply_message)
reply_message = reply_message
# 触发NormalMessageResponded事件
args = {
"launcher_type": kwargs['launcher_type'],
"launcher_id": kwargs['launcher_id'],
"sender_id": kwargs['sender_id'],
"session": None,
"prefix": prefix,
"response_text": reply_message,
"finish_reason": "revLibs."+revcfg.reverse_lib+".finish",
}
inter_event: EventContext = emit(NormalMessageResponded, **args)
reply = []
if inter_event.get_return_value("prefix") is not None:
prefix = inter_event.get_return_value("prefix")
if inter_event.get_return_value("reply") is not None:
reply = inter_event.get_return_value("reply")
if not inter_event.is_prevented_default():
reply = [prefix + reply_message]
except Exception as e:
logging.error("[rev] " + traceback.format_exc())
import config
if config.hide_exce_info_to_user:
reply_message = ''
try:
import tips
reply_message = tips.alter_tip_message
except:
if hasattr(config, "alter_tip_message"):
reply_message = config.alter_tip_message
else:
reply_message = "处理消息时出现错误，请联系管理员"
kwargs['host'].notify_admin("[rev] 处理消息时出现错误:\n"+traceback.format_exc())
else:
reply_message = "处理消息时出现错误，请联系管理员"+"\n"+traceback.format_exc()
reply = [revcfg.reply_prefix+reply_message]
if reply_message != "":
event.add_return(
"reply",
reply,
)
@on(PersonCommandSent)
@on(GroupCommandSent)
def command_send(inst, event: EventContext, **kwargs):
reply_message = ""
try:
reply_message = proccmd.process_command(session_name=kwargs['launcher_type']+"_"+str(kwargs['launcher_id']),
**kwargs)
logging.debug("[rev] " + reply_message)
except Exception as e:
logging.error("[rev] " + traceback.format_exc())
reply_message = "处理命令时出现错误，请联系管理员"+"\n"+traceback.format_exc()
if reply_message.strip() != "":
event.add_return(
"reply",
["{}(cmd)".format(revcfg.reply_prefix)+reply_message],
)
event.prevent_default()
event.prevent_postorder()
def make_reply(self, prompt, **kwargs) -> dict:
reply_gen = self.chatbot.ask(prompt, **kwargs)
reply = {}
for r in reply_gen:
reply = r
return reply
# 插件卸载时触发
def __del__(self):
pass