-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
98 lines (84 loc) · 3.06 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
"""
Discord ChatGPT bot
"""
import os
import logging
from typing import Annotated
import json
import uvicorn
import requests
import openai
from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError
from fastapi import FastAPI, Header, HTTPException, BackgroundTasks, Request
PUBLIC_KEY = os.getenv("APPLICATION_PUBLIC_KEY")
openai.organization = os.getenv("OPENAI_ORG")
openai.api_key = os.getenv("OPENAI_API_KEY")
verify_key = VerifyKey(bytes.fromhex(PUBLIC_KEY))
logging.basicConfig(level=logging.INFO)
app = FastAPI()
# write a path called /ping that returns "pong" used for a startup check
@app.get("/ping")
async def ping():
return "pong"
@app.post("/interaction")
async def interaction2(
background_tasks: BackgroundTasks,
request: Request,
x_signature_ed25519: Annotated[str, Header()] = None,
x_signature_timestamp: Annotated[str, Header()] = None,
):
request_body = await request.body()
decoded_body = json.loads(request_body.decode("utf-8"))
message = x_signature_timestamp.encode() + request_body
try:
verify_key.verify(message, bytes.fromhex(x_signature_ed25519))
# logging.info(f'Signature verification succeeded for {request_body}")
except (BadSignatureError, KeyError) as exc:
# logging.warning(f'Signature verification failed for {request_body}")
raise HTTPException(status_code=401) from exc
if decoded_body["type"] == 1:
return {"type": 1}
elif decoded_body["type"] == 2:
response_content = (f'Question from '
f'{decoded_body["member"]["user"]["username"]}: '
f'{decoded_body["data"]["options"][0]["value"]}')
response = {"type": 4, "data": {"content": response_content}}
openai_content = {
"token": decoded_body["token"],
"application_id": decoded_body["application_id"],
"signature": x_signature_ed25519,
"timestamp": x_signature_timestamp,
"orig_body": decoded_body,
"orig_data": decoded_body["data"]["options"][0]["value"]
}
background_tasks.add_task(check_openai, message=openai_content)
return response
return
def check_openai(message: str):
completion = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant, with a cheeky personality."},
{"role": "user", "content": message["orig_data"]}
])
result = completion.choices[0].message.content
url = (f'https://discord.com/api/v10/webhooks/'
f'{message["application_id"]}/{message["token"]}')
payload = {
"content": result
}
return requests.post(url, json=payload, timeout=20)
if __name__ == "__main__":
print("Starting webserver...")
uvicorn.run(
"main:app",
host="0.0.0.0",
port=int(os.getenv("PORT", "8080")),
log_level="info",
proxy_headers=True
# ssl_keyfile="./cert.key",
# ssl_certfile="./cert.cer",
# ssl_ca_certs="./ca.cer",
# reload=True
)