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
|
import random
from slack_sdk.web.async_client import AsyncWebClient
def generate_voting_blocks(deployment_id: int, message: str, approve_count: int, reject_count: int, voting_enabled: bool):
blocks = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"Voting | Deployment #{deployment_id}"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Message:*\n> {message}"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Votes:*\n> Approve: {approve_count}\n> Reject: {reject_count}"
}
},
]
if voting_enabled:
blocks.append({
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "Approve ^_^"},
"style": "primary",
"action_id": f"vote_approve_{deployment_id}"
},
{
"type": "button",
"text": {"type": "plain_text", "text": "Reject T_T"},
"style": "danger",
"action_id": f"vote_reject_{deployment_id}"
}
]
})
return blocks
async def get_random_channel(oauth_token: str):
client = AsyncWebClient(token=oauth_token)
try:
response = await client.conversations_list(types="public_channel", limit=200, exclude_archived=True)
channels = [c for c in response.get("channels", []) if c.get("is_member", False)]
if not channels:
return None
channel_approved = False
while not channel_approved:
channel = random.choice(channels)
# FIXME: Check if CM-approved here
channel_approved = True
return channel
except Exception as e:
print(f"random channel is broken, we might be cooked: {e}")
return None
|
