/
main.py
100 lines (81 loc) · 2.78 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
from typing import Optional
from functools import lru_cache
import logging
import os
from fastapi import BackgroundTasks, Depends, FastAPI, Form, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
import requests
import config
logging.basicConfig(filename="logging.conf")
@lru_cache()
def get_settings():
return config.Settings()
app = FastAPI()
templates = Jinja2Templates(directory="templates")
class SlackEvent(BaseModel):
type: str
channel: str
user: str
ts: str
thread_ts: Optional[str]
channel_type: str
text: str
class SlackRequest(BaseModel):
token: str
challenge: Optional[str]
type: str
event: SlackEvent
def is_sender_a_bot(event: SlackEvent) -> bool:
params_payload = {
"user": event.user,
}
headers_payload = {
"Authorization": f"Bearer {get_settings().slack_oauth_token}"
}
response = requests.get("https://slack.com/api/users.info", params=params_payload, headers=headers_payload)
if response.ok:
return response.json().get("user", {}).get("is_bot", True)
return True
async def respond_in_thread(event: SlackEvent):
is_response_required = True
if is_sender_a_bot(event):
is_response_required = False
if event.thread_ts:
is_response_required = False
if not is_response_required:
return
message = None
with open(get_settings().message_file_name, 'r') as file:
message = file.read()
payload = {
"token": get_settings().slack_oauth_token,
"text": message,
"channel": event.channel,
"thread_ts": event.ts
}
r = requests.post('https://slack.com/api/chat.postMessage', data=payload)
log_message = f"Status: {r.status_code}\nBody: {r.json()}"
logging.info(log_message)
@app.post("/editMessage")
async def handle_edit_message(message: str = Form(...), settings: config.Settings = Depends(get_settings)):
with open(settings.message_file_name, 'w+') as file:
file.write(message)
return {"message": message}
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
logging.info(str(request))
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/")
async def handle_slack_event(request: SlackRequest, background_tasks: BackgroundTasks):
logging.info(request)
print(request)
type = request.type
if request.type == "url_verification":
return get_challenge_response(request)
elif request.type == "event_callback":
background_tasks.add_task(respond_in_thread, request.event)
async def get_challenge_response(request: SlackRequest):
background_tasks.add_task(get_identity)
return {"challenge": request.challenge}