CTFeed/src/utils/join_channel.py at a9561291dadfb56052357a0da4178d50e9d00eac · pictures2333/CTFeed · GitHub

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
128
129
130
131
132
133
134
135
136
137
138
from typing import Optional
import logging

from discord.ext import commands
import discord

from src.database.model import Event
from src.database.database import get_db
from src import crud
from src.utils.ctf_api import fetch_ctf_events
from src.utils.embed_creator import create_event_embed
from src.config import settings

logger = logging.getLogger(__name__)

async def join_channel(
    bot:commands.Bot,
    interaction:discord.Interaction,
    event_id:int,
):
    await interaction.response.defer(ephemeral=True)

    async with get_db() as session:
        # get event from database
        events = await crud.read_event(session, event_id=[event_id])
        if len(events) != 1:
            await interaction.followup.send(content="Invalid event", ephemeral=True)
            return
        event:Event = events[0]

        if not(event.channel_id is None) and not(bot.get_channel(event.channel_id) is None): # channel exists
            try:
                member = interaction.guild.get_member(interaction.user.id)
                channel = bot.get_channel(event.channel_id)

                if channel.permissions_for(member).view_channel == True:
                    await interaction.followup.send(content="You have joined the channel", ephemeral=True)
                    return

                await channel.set_permissions(member, view_channel=True)
                await channel.send(embed=discord.Embed(
                    color=discord.Color.green(),
                    title=f"{interaction.user.display_name} joined the channel"
                ))

                await interaction.followup.send(content="Done", ephemeral=True)

                logger.info(f"User {interaction.user.display_name}(id={interaction.user.id}) joined channel {channel.name}(id={channel.id})")
                return
            except Exception as e:
                logger.error(f"Failed to join channel: {e}")
                await interaction.followup.send(content=f"Failed to join channel: {e}", ephemeral=True)
                return
        else: # channel not found or invalid
            # get event from CTFTime
            events_api = await fetch_ctf_events(event.event_id)
            if len(events_api) != 1:
                await interaction.followup.send(content="Invalid event", ephemeral=True)
                return
            event_api:Event = events_api[0]

            # create channel
            category_id = settings.CTF_CHANNEL_CATETORY_ID
            guild = interaction.guild
            category = discord.utils.get(interaction.guild.categories, id=category_id)
            if category is None:
                await interaction.followup.send(content=f"Category id={category_id} not found", ephemeral=True)
                return

            overwrites = {
                guild.default_role: discord.PermissionOverwrite(view_channel=False),
                interaction.user: discord.PermissionOverwrite(view_channel=True),
                guild.me: discord.PermissionOverwrite(view_channel=True)
            }

            try:
                new_channel = await guild.create_text_channel(event.title, category=category, overwrites=overwrites)
                event_id = event.event_id
                event = await crud.update_event(session, event_id=event_id, channel_id=new_channel.id)
                if event is None:
                    await new_channel.delete(reason=f"Failed to update event_id={event_id} on database")
                    await interaction.followup.send(content=f"Failed to create channel: Failed to update event_id={event_id} on database", ephemeral=True)
                    return

                embed = await create_event_embed(event_api, f"{interaction.user.display_name} 發起了 {event.title}")

                await new_channel.send(embed=embed)

                await interaction.followup.send(content="Done", ephemeral=True)

                logger.info(f"User {interaction.user.display_name}(id={interaction.user.id}) created and joined channel {new_channel.name}(id={new_channel.id})")
                return
            except Exception as e:
                logger.error(f"Failed to create channel: {e}")
                await interaction.followup.send(content=f"Failed to create channel: {e}", ephemeral=True)
                return


async def join_channel_custom(
    bot:commands.Bot,
    interaction:discord.Interaction,
    channel_id:int,
):
    await interaction.response.defer(ephemeral=True)

    # get channel from database
    async with get_db() as session:
        channel_db = await crud.read_custom_channel(session, channel_id=channel_id)
    if len(channel_db) != 1:
        await interaction.followup.send(f"Channel (id={channel_id}) not found", ephemeral=True)
        return

    # get channel from discord
    channel = bot.get_channel(channel_id)
    if channel is None:
        await interaction.followup.send(f"Channel (id={channel_id}) not found", ephemeral=True)
        return

    # join channel
    try:
        if channel.permissions_for(interaction.user).view_channel == True:
            await interaction.followup.send(content="You have joined the channel", ephemeral=True)
            return

        await channel.set_permissions(interaction.user, view_channel=True)
        await channel.send(embed=discord.Embed(
            color=discord.Color.green(),
            title=f"{interaction.user.display_name} joined the channel"
        ))

        await interaction.followup.send(content="Done", ephemeral=True)

        logger.info(f"User {interaction.user.display_name}(id={interaction.user.id}) joined channel {channel.name}(id={channel.id})")
        return
    except Exception as e:
        logger.error(f"Failed to join channel: {e}")
        await interaction.followup.send(content=f"Failed to join channel: {e}", ephemeral=True)
        return