Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 84 additions & 64 deletions announcement/announcement.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import json
from pathlib import Path
from typing import Optional, TYPE_CHECKING
from typing import TYPE_CHECKING

import discord

Expand All @@ -11,14 +11,16 @@
from core import checks
from core.models import getLogger, PermissionLevel

from .core.models import AnnouncementModel
from .core.models import AnnouncementModel, AnnouncementType
from .core.views import AnnouncementView


if TYPE_CHECKING:
from bot import ModmailBot


logger = getLogger(__name__)

info_json = Path(__file__).parent.resolve() / "info.json"
with open(info_json, encoding="utf-8") as f:
__plugin_info__ = json.loads(f.read())
Expand All @@ -27,7 +29,35 @@
__description__ = "\n".join(__plugin_info__["description"]).format(__version__)


logger = getLogger(__name__)
news_channel_hyperlink = (
"[announcement](https://support.discord.com/hc/en-us/articles/360032008192-Announcement-Channels)"
)
type_desc = (
"Choose a type of announcement.\n\n"
"__**Available types:**__\n"
"- **Plain** : Plain text announcement.\n"
"- **Embed** : Embedded announcement. Image and thumbnail image are also supported.\n"
)
embed_desc = (
"Click the `Edit` button below to set/edit the embed values.\n\n"
"__**Available fields:**__\n"
"- **Description** : The content of the announcement. Must not exceed 4000 characters.\n"
"- **Thumbnail URL** : URL of the image shown at the top right of the embed.\n"
"- **Image URL** : URL of the large image shown at the bottom of the embed.\n"
"- **Color** : The color code of the embed. If not specified, fallbacks to bot main color. "
"The following formats are accepted:\n - `0x<hex>`\n - `#<hex>`\n - `0x#<hex>`\n - `rgb(<number>, <number>, <number>)`\n"
"Like CSS, `<number>` can be either 0-255 or 0-100% and `<hex>` can be either a 6 digit hex number or a 3 digit hex shortcut (e.g. #fff).\n"
)
plain_desc = "Click the `Edit` button below to set/edit the content.\n"
mention_desc = (
"If nothing is selected, the announcement will be posted without any mention.\n"
"To mention Users or Roles, select `Others` in the first dropdown, then in second dropdown select Users or Roles you want to mention.\n"
)
channel_desc = (
"The destination channel. If nothing is selected, the announcement will be posted "
"in the current channel.\n"
f"The announcement can be published if the type of destination channel is {news_channel_hyperlink} channel.\n"
)


class Announcement(commands.Cog):
Expand All @@ -40,80 +70,64 @@ def __init__(self, bot: ModmailBot):
@checks.has_permissions(PermissionLevel.ADMINISTRATOR)
async def announce(self, ctx: commands.Context):
"""
Base command to create announcements.
"""
await ctx.send_help(ctx.command)
Post an announcement.

@announce.command(name="create", aliases=["start"])
@checks.has_permissions(PermissionLevel.ADMINISTRATOR)
async def announce_create(self, ctx: commands.Context, *, channel: Optional[discord.TextChannel] = None):
Run this command without argument to initiate a creation panel where you can choose and customise the output of the announcement.
"""
Post an announcement in a channel specified.
announcement = AnnouncementModel(ctx)
sessions = [
("type", type_desc),
("embed", embed_desc),
("plain", plain_desc),
("mention", mention_desc),
("channel", channel_desc),
("publish", None),
]
view = AnnouncementView(ctx, announcement, input_sessions=sessions)
await view.create_base()

This will initiate a creation panel where you can choose and customise the output of the announcement.
await view.wait()
if not announcement.is_ready():
# cancelled or timed out
return

`channel` if specified may be a channel ID, mention, or name. Otherwise, fallbacks to current channel.
await announcement.send()

__**Note:**__
- If `channel` is not specified, to ensure cleaner output the creation message will automatically be deleted after the announcement is posted.
"""
delete = False
if channel is None:
channel = ctx.channel
delete = True
if announcement.channel == ctx.channel:
view.stop()
try:
await ctx.message.delete()
await view.message.delete()
except discord.Forbidden:
logger.warning(f"Missing `Manage Messages` permission in {channel} channel.")

announcement = AnnouncementModel(ctx, channel)
view = AnnouncementView(ctx, announcement)
embed = discord.Embed(title="Announcement Creation Panel", color=self.bot.main_color)
embed.description = (
"Choose a type of announcement using the dropdown menu below.\n\n"
"__**Available types:**__\n"
"- **Normal** : Plain text announcement.\n"
"- **Embed** : Embedded announcement. Image and thumbnail image are also supported."
)
view.message = message = await ctx.send(embed=embed, view=view)
await view.wait(input_event=True)

if not announcement.posted:
return

if delete:
view.stop()
await message.delete()
logger.warning(f"Missing `Manage Messages` permission in {ctx.channel} channel.")
return

embed = message.embeds[0]
description = f"Announcement has been posted in {channel.mention}.\n\n"
embed = view.message.embeds[0]
description = f"Announcement has been posted in {announcement.channel.mention}.\n\n"
if announcement.channel.type == discord.ChannelType.news:
description += "Would you like to publish this announcement?\n\n"
view.generate_buttons(confirmation=True)
description += "Would you like to publish the announcement?\n\n"
view.fill_items(confirmation=True)
else:
view.stop()

embed.description = description
await message.edit(embed=embed, view=view)
await view.message.edit(embed=embed, view=view)
if view.is_finished():
return

await view.wait()

hyper_link = f"[announcement]({announcement.message.jump_url})"
if view.confirm:
await announcement.publish()
embed.description = f"Successfully published this {hyper_link} to all subscribed channels.\n\n"
if view.confirm is not None:
if not view.confirm:
if view.confirmed is not None:
hyper_link = f"[announcement]({announcement.message.jump_url})"
if view.confirmed:
await announcement.publish()
embed.description = f"Successfully published this {hyper_link} to all following servers.\n\n"
else:
embed.description = (
f"To manually publish this {hyper_link}, use command:\n"
f"```\n{ctx.prefix}publish {announcement.channel.id}-{announcement.message.id}\n```"
)
view = None

await message.edit(embed=embed, view=view)
await view.message.edit(embed=embed, view=None)

@announce.command(name="quick")
@checks.has_permissions(PermissionLevel.ADMINISTRATOR)
Expand All @@ -123,19 +137,25 @@ async def announce_quick(self, ctx: commands.Context, channel: discord.TextChann

`channel` may be a channel ID, mention, or name.
"""
await channel.send(content)

@commands.command()
announcement = AnnouncementModel(ctx, type=AnnouncementType.PLAIN, channel=channel, content=content)
await announcement.send()

@commands.command(
help=(
"Publish a message from announcement channel to all channels in other servers that are "
"following the channel.\n\n"
"`message` may be a message ID, format of `channel_id`-`message_id` "
"(e.g. `1079077919915266210-1079173422967439360`), or message link.\n\n"
"__**Notes:**__\n"
"- If message ID is provided (without channel ID and not the message link), the bot will only "
"look for the message in the current channel.\n"
f"- Only messages in {news_channel_hyperlink} channels can be published."
),
)
@checks.has_permissions(PermissionLevel.ADMINISTRATOR)
async def publish(self, ctx: commands.Context, *, message: discord.Message):
"""
Publish a message from announcement channel to all subscribed channels.

`message` may be a message ID, format of `channel ID-message ID`, or message link.

__**Notes:**__
- If message ID is provided (without channel ID and not the message link), the bot will only look for the message in the current channel.
- Only messages in [announcement](https://support.discord.com/hc/en-us/articles/360032008192-Announcement-Channels) channels can be published.
Publish a message from announcement channel.
"""
channel = message.channel
if not channel.type == discord.ChannelType.news:
Expand All @@ -145,7 +165,7 @@ async def publish(self, ctx: commands.Context, *, message: discord.Message):

await message.publish()
embed = discord.Embed(
description=f"Successfully published this [message]({message.jump_url}) to all subscribed channels.",
description=f"Successfully published this [message]({message.jump_url}) to all following servers.",
color=self.bot.main_color,
)
await ctx.reply(embed=embed)
Expand Down
97 changes: 51 additions & 46 deletions announcement/core/models.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
import asyncio

from enum import Enum
from typing import Any, Dict, Optional
from typing import Any, Dict

import discord
from discord.utils import MISSING
from discord.ext import commands


__all__ = ("AnnouncementType", "AnnouncementModel")


def _color_converter(value: str) -> int:
try:
return int(value)
Expand All @@ -21,7 +24,7 @@ def _color_converter(value: str) -> int:

class AnnouncementType(Enum):
# only two are valid for now. may add more later.
NORMAL = "normal"
PLAIN = "plain"
EMBED = "embed"
INVALID = "invalid"

Expand All @@ -45,59 +48,54 @@ def value(self) -> str:


class AnnouncementModel:
def __init__(self, ctx: commands.Context, channel: discord.TextChannel):
"""
Represents an instance to manage announcement creation.
"""

def __init__(
self,
ctx: commands.Context,
*,
type: AnnouncementType = MISSING,
channel: discord.TextChannel = MISSING,
content: str = MISSING,
embed: discord.Embed = MISSING,
):
self.ctx: commands.Context = ctx
self.type: AnnouncementType = type
self.channel: discord.TextChannel = channel
self.event: asyncio.Event = asyncio.Event()
self.ready: bool = False
self.content: str = content
self.embed: discord.Embed = embed

self.type: AnnouncementType = MISSING
self.message: discord.Message = MISSING
self.content: str = MISSING
self.embed: discord.Embed = MISSING
self.event: asyncio.Event = asyncio.Event()
self.ready: bool = False
self.task: asyncio.Task = MISSING

@property
def posted(self) -> bool:
return self.event.is_set()
def is_ready(self) -> bool:
"""
Returns whether the announcement is ready to be posted.
"""
return self.ready and self.event.is_set()

@posted.setter
def posted(self, flag: bool) -> None:
if flag:
self.event.set()
else:
if self.task is not MISSING:
self.task.cancel()
self.event.clear()
def cancel(self) -> None:
"""Cancel the announcement."""
self.ready = False
if self.task is not MISSING:
self.task.cancel()
self.event.clear()

async def wait(self) -> None:
self.task = self.ctx.bot.loop.create_task(self.event.wait())
"""
Wait until the announcement is ready to be posted or cancelled.
"""
if self.task is MISSING:
self.task = self.ctx.bot.loop.create_task(self.event.wait())
try:
await self.task
except asyncio.CancelledError:
pass

async def resolve_mentions(self) -> None:
if not self.content:
return
ret = []
argument = self.content.split()
for arg in argument:
if arg in ("@here", "@everyone"):
ret.append(arg)
continue
user_or_role = None
try:
user_or_role = await commands.RoleConverter().convert(self.ctx, arg)
except commands.BadArgument:
try:
user_or_role = await commands.MemberConverter().convert(self.ctx, arg)
except commands.BadArgument:
raise commands.BadArgument(f"Unable to convert {arg} to user or role mention.")
if user_or_role is not None:
ret.append(user_or_role.mention)
self.content = ", ".join(ret) if ret else None

def create_embed(
self,
*,
Expand All @@ -106,6 +104,9 @@ def create_embed(
thumbnail_url: str = MISSING,
image_url: str = MISSING,
) -> discord.Embed:
"""
Create the announcement embed.
"""
if not color:
color = self.ctx.bot.main_color
else:
Expand All @@ -117,7 +118,7 @@ def create_embed(
embed.set_thumbnail(url=thumbnail_url)
if image_url:
embed.set_image(url=image_url)
embed.set_footer(text="Announcement", icon_url=self.channel.guild.icon)
embed.set_footer(text="Announcement", icon_url=self.ctx.guild.icon)
self.embed = embed
return embed

Expand All @@ -127,13 +128,17 @@ def send_params(self) -> Dict[str, Any]:
params["content"] = self.content
return params

async def post(self) -> None:
async def send(self) -> None:
"""
Send the announcement message.
"""
if not self.channel:
self.channel = self.ctx.channel
self.message = await self.channel.send(**self.send_params())
self.posted = True

async def publish(self) -> None:
"""
Publishes the announcement. This will only work if the channel type is a news channel
and if the announcement has not been posted yet.
Publish the announcement. This will only work if the channel type is a news channel
and if the announcement has never been published yet.
"""
await self.message.publish()
Loading