-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add a utils to write data to file asynchronously (for logging the anonymous messages)
- Loading branch information
Showing
5 changed files
with
81 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,52 @@ | ||
from __future__ import annotations | ||
|
||
from datetime import datetime | ||
from pathlib import Path | ||
from typing import TYPE_CHECKING | ||
|
||
from discord import Interaction, TextChannel, app_commands | ||
from discord.ext import commands | ||
|
||
from etuutt_bot.utils.data_write import data_write | ||
|
||
if TYPE_CHECKING: | ||
from etuutt_bot.bot import EtuUTTBot | ||
|
||
|
||
class AnonMsgCog(commands.GroupCog, group_name="anon"): | ||
"""Commandes pour envoyer des messages anonymes.""" | ||
|
||
def __init__(self, bot: EtuUTTBot): | ||
self.bot = bot | ||
|
||
@app_commands.command(name="send") | ||
@app_commands.rename(channel="salon") | ||
async def send(self, interaction: Interaction[EtuUTTBot], channel: TextChannel, message: str): | ||
"""Envoie un message anonyme dans le salon demandé s'il le supporte.""" | ||
if channel.id not in self.bot.settings.guild.anonymous_channels: | ||
await interaction.response.send_message( | ||
"Le salon ne supporte pas les messages anonymes" | ||
) | ||
return | ||
msg = await channel.send(f"[ANONYME] {message}") | ||
await data_write( | ||
f"[{datetime.now().isoformat(' ', 'seconds')}]" | ||
f" {interaction.user.name} alias {interaction.user.display_name}" | ||
f" a envoyé le message {msg.id} sur le salon #{channel.name}.\n", | ||
Path("data", "logs", "anon.log"), | ||
) | ||
await interaction.response.send_message( | ||
f"Votre message a bien été envoyé : {msg.jump_url}", ephemeral=True | ||
) | ||
|
||
@app_commands.command(name="liste") | ||
async def list(self, interaction: Interaction[EtuUTTBot]): | ||
"""Envoie la liste des salons dans lesquels un message anonyme peut être envoyé.""" | ||
if not self.bot.settings.guild.anonymous_channels: | ||
await interaction.response.send_message( | ||
"Il n'y a pas de salon configuré pour envoyer un message anonyme." | ||
) | ||
msg = "Les salons dans lesquels un message anonyme peut être envoyé sont :" | ||
for c in self.bot.settings.guild.anonymous_channels: | ||
msg += f"\n- {interaction.guild.get_channel(c).mention}" | ||
await interaction.response.send_message(msg) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,25 @@ | ||
from asyncio import Lock, to_thread | ||
from pathlib import Path | ||
|
||
FILE_LOCKS = {} | ||
|
||
|
||
def _get_lock(file: Path) -> Lock: | ||
"""Retourne le verrou associé à un chemin et le crée s'il n'existe pas.""" | ||
for path, lock in FILE_LOCKS.items(): | ||
if path == file: | ||
return lock | ||
FILE_LOCKS.update({file: Lock()}) | ||
return FILE_LOCKS.get(file) | ||
|
||
|
||
def _file_write(data: str, file: Path) -> None: | ||
"""Ajoute une chaîne de caractères à la fin d'un fichier.""" | ||
with open(file, "a") as f: | ||
f.write(data) | ||
|
||
|
||
async def data_write(data: str, file: Path) -> None: | ||
"""Dans un autre thread, écrit une chaîne de caractères dans un fichier.""" | ||
async with _get_lock(file): | ||
await to_thread(_file_write, data, file) |