-
Notifications
You must be signed in to change notification settings - Fork 4
/
blackbox.py
127 lines (110 loc) · 4.71 KB
/
blackbox.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
import uuid
import re
from aiohttp import ClientSession, FormData
from pyrogram import Client, filters
from utils.misc import modules_help, prefix
def id_generator() -> str:
return str(uuid.uuid4())
@Client.on_message(filters.command(["bbox", "blackbox"], prefix) & filters.me)
async def blackbox(client, message):
m = message
msg = await m.edit_text("🔍")
if len(m.text.split()) == 1:
return await msg.edit_text(
"Type some query buddy 🐼\n"
f"{prefix}blackbox text with reply to the photo or just text"
)
else:
try:
session = ClientSession()
prompt = m.text.split(maxsplit=1)[1]
user_id = id_generator()
image = None
if m.reply_to_message and (
m.reply_to_message.photo
or (m.reply_to_message.sticker and not m.reply_to_message.sticker.is_video)
):
file_name = f"blackbox_{m.chat.id}.jpeg"
file_path = await m.reply_to_message.download(file_name=file_name)
with open(file_path, "rb") as file:
image = file.read()
if image:
data = FormData()
data.add_field("fileName", file_name)
data.add_field("userId", user_id)
data.add_field(
"image", image, filename=file_name, content_type="image/jpeg"
)
api_url = "https://www.blackbox.ai/api/upload"
try:
async with session.post(api_url, data=data) as response:
response_json = await response.json()
except Exception as e:
return await msg.edit(f"❌ Error: {str(e)}")
messages = [
{
"role": "user",
"content": response_json["response"] + "\n#\n" + prompt,
}
]
data = {
"messages": messages,
"user_id": user_id,
"codeModelMode": True,
"agentMode": {},
"trendingAgentMode": {},
}
headers = {"Content-Type": "application/json"}
url = "https://www.blackbox.ai/api/chat"
try:
async with session.post(url, headers=headers, json=data) as response:
response_text = await response.text()
except Exception as e:
return await msg.edit(f"❌ Error: {str(e)}")
cleaned_response_text = re.sub(
r"^\$?@?\$?v=undefined-rv\d+@?\$?|\$?@?\$?v=v\d+\.\d+-rv\d+@?\$?",
"",
response_text,
)
text = cleaned_response_text.strip()[2:]
if "$~~~$" in text:
text = re.sub(r"\$~~~\$.*?\$~~~\$", "", text, flags=re.DOTALL)
rdata = {"reply": text}
return await msg.edit_text(text=rdata["reply"])
else:
reply = m.reply_to_message
if reply and reply.text:
prompt = f"Old conversation:\n{reply.text}\n\nQuestion:\n{prompt}"
messages = [{"role": "user", "content": prompt}]
data = {
"messages": messages,
"user_id": user_id,
"codeModelMode": True,
"agentMode": {},
"trendingAgentMode": {},
}
headers = {"Content-Type": "application/json"}
url = "https://www.blackbox.ai/api/chat"
try:
async with session.post(url, headers=headers, json=data) as response:
response_text = await response.text()
except Exception as e:
return await msg.edit(f"❌ Error: {str(e)}")
cleaned_response_text = re.sub(
r"^\$?@?\$?v=undefined-rv\d+@?\$?|\$?@?\$?v=v\d+\.\d+-rv\d+@?\$?",
"",
response_text,
)
text = cleaned_response_text.strip()[2:]
if "$~~~$" in text:
text = re.sub(r"\$~~~\$.*?\$~~~\$", "", text, flags=re.DOTALL)
rdata = {"reply": text}
return await msg.edit_text(text=rdata["reply"])
except Exception as e:
return await msg.edit(f"�� Error: {str(e)}")
finally:
await session.close()
modules_help["blackbox"] = {
"blackbox [query]*": "Ask anything to Blackbox",
"bbox [query]*": "Ask anything to Blackbox",
}