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
|
import random
import httpx
from slack_sdk.web.async_client import AsyncWebClient
from .db import get_approved_channels
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(user_oauth_token: str, user_id: str) -> str | None:
approved = await get_approved_channels()
print(f"DEBUG get_random_channel: approved = {approved}")
if not approved:
print("DEBUG get_random_channel: No approved channels")
return None
random.shuffle(approved)
print(f"DEBUG get_random_channel: shuffled approved = {approved}")
for channel_id in approved:
print(f"DEBUG get_random_channel: checking channel {channel_id} for user {user_id}")
is_member = await user_is_in_channel(user_oauth_token, channel_id, user_id)
print(f"DEBUG get_random_channel: is_member = {is_member}")
if is_member:
print(f"DEBUG get_random_channel: FOUND! returning {channel_id}")
return channel_id
print("DEBUG get_random_channel: No channels found")
return None
async def user_is_in_channel(oauth_token: str, channel_id: str, user_id: str) -> bool:
client = AsyncWebClient(token=oauth_token)
try:
print(f"DEBUG user_is_in_channel: calling conversations_members for {channel_id}")
response = await client.conversations_members(channel=channel_id)
print(f"DEBUG user_is_in_channel: response ok = {response.get('ok')}")
members = response.get("members", [])
print(f"DEBUG user_is_in_channel: members count = {len(members)}")
print(f"DEBUG user_is_in_channel: looking for {user_id} in members")
return user_id in members
except Exception as e:
print(f"DEBUG user_is_in_channel: EXCEPTION - {e}")
return False
async def is_user_cm(channel_id: str, user_id: str):
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"https://flaron.halceon.dev/channel/{channel_id}")
if response.status_code != 200:
return False
data = response.json()
managers = data.get("managers", [])
return user_id in managers
except Exception in e:
print(f"flaron api failed T_T {e}")
return False
|
