Merge branch 'master' into cogguide_develop
This commit is contained in:
bobloy
commit
bd45f353a8
@ -54,8 +54,7 @@ class AnnounceDaily(Cog):
|
||||
|
||||
Do `[p]help annd <subcommand>` for more details
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@commands.command()
|
||||
@checks.guildowner()
|
||||
|
||||
@ -168,7 +168,7 @@ class AudioTrivia(Trivia):
|
||||
@commands.guild_only()
|
||||
async def audiotrivia_list(self, ctx: commands.Context):
|
||||
"""List available trivia including audio categories."""
|
||||
lists = set(p.stem for p in self._all_audio_lists())
|
||||
lists = {p.stem for p in self._all_audio_lists()}
|
||||
if await ctx.embed_requested():
|
||||
await ctx.send(
|
||||
embed=discord.Embed(
|
||||
|
||||
@ -48,8 +48,7 @@ class CCRole(commands.Cog):
|
||||
"""Custom commands management with roles
|
||||
|
||||
Highly customizable custom commands with role management."""
|
||||
if not ctx.invoked_subcommand:
|
||||
pass
|
||||
pass
|
||||
|
||||
@ccrole.command(name="add")
|
||||
@checks.mod_or_permissions(administrator=True)
|
||||
@ -228,7 +227,7 @@ class CCRole(commands.Cog):
|
||||
if not role_list:
|
||||
return "None"
|
||||
return ", ".join(
|
||||
[discord.utils.get(ctx.guild.roles, id=roleid).name for roleid in role_list]
|
||||
discord.utils.get(ctx.guild.roles, id=roleid).name for roleid in role_list
|
||||
)
|
||||
|
||||
embed.add_field(name="Text", value="```{}```".format(cmd["text"]), inline=False)
|
||||
@ -252,7 +251,7 @@ class CCRole(commands.Cog):
|
||||
)
|
||||
return
|
||||
|
||||
cmd_list = ", ".join([ctx.prefix + c for c in sorted(cmd_list.keys())])
|
||||
cmd_list = ", ".join(ctx.prefix + c for c in sorted(cmd_list.keys()))
|
||||
cmd_list = "Custom commands:\n\n" + cmd_list
|
||||
|
||||
if (
|
||||
@ -325,9 +324,7 @@ class CCRole(commands.Cog):
|
||||
|
||||
async def eval_cc(self, cmd, message: discord.Message, ctx: commands.Context):
|
||||
"""Does all the work"""
|
||||
if cmd["proles"] and not (
|
||||
set(role.id for role in message.author.roles) & set(cmd["proles"])
|
||||
):
|
||||
if cmd["proles"] and not {role.id for role in message.author.roles} & set(cmd["proles"]):
|
||||
log.debug(f"{message.author} missing required role to execute {ctx.invoked_with}")
|
||||
return # Not authorized, do nothing
|
||||
|
||||
|
||||
@ -59,6 +59,35 @@ Install these on your windows machine before attempting the installation:
|
||||
[Pandoc - Universal Document Converter](https://pandoc.org/installing.html)
|
||||
|
||||
## Methods
|
||||
### Automatic
|
||||
|
||||
This method requires some luck to pull off.
|
||||
|
||||
#### Step 1: Add repo and install cog
|
||||
|
||||
```
|
||||
[p]repo add Fox https://github.com/bobloy/Fox-V3
|
||||
[p]cog install Fox chatter
|
||||
```
|
||||
|
||||
If you get an error at this step, stop and skip to one of the manual methods below.
|
||||
|
||||
#### Step 2: Install additional dependencies
|
||||
|
||||
Assuming the previous commands had no error, you can now use `pipinstall` to add the remaining dependencies.
|
||||
|
||||
NOTE: This method is not the intended use case for `pipinstall` and may stop working in the future.
|
||||
|
||||
```
|
||||
[p]pipinstall --no-deps chatterbot>=1.1
|
||||
```
|
||||
|
||||
#### Step 3: Load the cog and get started
|
||||
|
||||
```
|
||||
[p]load chatter
|
||||
```
|
||||
|
||||
### Windows - Manually
|
||||
#### Step 1: Built-in Downloader
|
||||
|
||||
|
||||
329
chatter/chat.py
329
chatter/chat.py
@ -2,8 +2,10 @@ import asyncio
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional
|
||||
from functools import partial
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import discord
|
||||
from chatterbot import ChatBot
|
||||
@ -15,6 +17,8 @@ from redbot.core.commands import Cog
|
||||
from redbot.core.data_manager import cog_data_path
|
||||
from redbot.core.utils.predicates import MessagePredicate
|
||||
|
||||
from chatter.trainers import MovieTrainer, TwitterCorpusTrainer, UbuntuCorpusTrainer2
|
||||
|
||||
log = logging.getLogger("red.fox_v3.chatter")
|
||||
|
||||
|
||||
@ -52,8 +56,14 @@ class Chatter(Cog):
|
||||
super().__init__()
|
||||
self.bot = bot
|
||||
self.config = Config.get_conf(self, identifier=6710497116116101114)
|
||||
default_global = {}
|
||||
default_guild = {"whitelist": None, "days": 1, "convo_delta": 15, "chatchannel": None}
|
||||
default_global = {"learning": True}
|
||||
default_guild = {
|
||||
"whitelist": None,
|
||||
"days": 1,
|
||||
"convo_delta": 15,
|
||||
"chatchannel": None,
|
||||
"reply": True,
|
||||
}
|
||||
path: pathlib.Path = cog_data_path(self)
|
||||
self.data_path = path / "database.sqlite3"
|
||||
|
||||
@ -73,6 +83,11 @@ class Chatter(Cog):
|
||||
|
||||
self.loop = asyncio.get_event_loop()
|
||||
|
||||
self._guild_cache = defaultdict(dict)
|
||||
self._global_cache = {}
|
||||
|
||||
self._last_message_per_channel: Dict[Optional[discord.Message]] = defaultdict(lambda: None)
|
||||
|
||||
async def red_delete_data_for_user(self, **kwargs):
|
||||
"""Nothing to delete"""
|
||||
return
|
||||
@ -81,7 +96,8 @@ class Chatter(Cog):
|
||||
|
||||
return ChatBot(
|
||||
"ChatterBot",
|
||||
storage_adapter="chatterbot.storage.SQLStorageAdapter",
|
||||
# storage_adapter="chatterbot.storage.SQLStorageAdapter",
|
||||
storage_adapter="chatter.storage_adapters.MyDumbSQLStorageAdapter",
|
||||
database_uri="sqlite:///" + str(self.data_path),
|
||||
statement_comparison_function=self.similarity_algo,
|
||||
response_selection_method=get_random_response,
|
||||
@ -91,7 +107,7 @@ class Chatter(Cog):
|
||||
logger=log,
|
||||
)
|
||||
|
||||
async def _get_conversation(self, ctx, in_channel: discord.TextChannel = None):
|
||||
async def _get_conversation(self, ctx, in_channels: List[discord.TextChannel]):
|
||||
"""
|
||||
Compiles all conversation in the Guild this bot can get it's hands on
|
||||
Currently takes a stupid long time
|
||||
@ -105,20 +121,12 @@ class Chatter(Cog):
|
||||
return msg.clean_content
|
||||
|
||||
def new_conversation(msg, sent, out_in, delta):
|
||||
# if sent is None:
|
||||
# return False
|
||||
|
||||
# Don't do "too short" processing here. Sometimes people don't respond.
|
||||
# if len(out_in) < 2:
|
||||
# return False
|
||||
|
||||
# print(msg.created_at - sent)
|
||||
|
||||
# Should always be positive numbers
|
||||
return msg.created_at - sent >= delta
|
||||
|
||||
for channel in ctx.guild.text_channels:
|
||||
if in_channel:
|
||||
channel = in_channel
|
||||
for channel in in_channels:
|
||||
# if in_channel:
|
||||
# channel = in_channel
|
||||
await ctx.maybe_send_embed("Gathering {}".format(channel.mention))
|
||||
user = None
|
||||
i = 0
|
||||
@ -153,11 +161,16 @@ class Chatter(Cog):
|
||||
except discord.HTTPException:
|
||||
pass
|
||||
|
||||
if in_channel:
|
||||
break
|
||||
# if in_channel:
|
||||
# break
|
||||
|
||||
return out
|
||||
|
||||
def _train_twitter(self, *args, **kwargs):
|
||||
trainer = TwitterCorpusTrainer(self.chatbot)
|
||||
trainer.train(*args, **kwargs)
|
||||
return True
|
||||
|
||||
def _train_ubuntu(self):
|
||||
trainer = UbuntuCorpusTrainer(
|
||||
self.chatbot, ubuntu_corpus_data_directory=cog_data_path(self) / "ubuntu_data"
|
||||
@ -165,6 +178,30 @@ class Chatter(Cog):
|
||||
trainer.train()
|
||||
return True
|
||||
|
||||
async def _train_movies(self):
|
||||
trainer = MovieTrainer(self.chatbot, cog_data_path(self))
|
||||
return await trainer.asynctrain()
|
||||
|
||||
async def _train_ubuntu2(self, intensity):
|
||||
train_kwarg = {}
|
||||
if intensity == 196:
|
||||
train_kwarg["train_dialogue"] = False
|
||||
train_kwarg["train_196"] = True
|
||||
elif intensity == 301:
|
||||
train_kwarg["train_dialogue"] = False
|
||||
train_kwarg["train_301"] = True
|
||||
elif intensity == 497:
|
||||
train_kwarg["train_dialogue"] = False
|
||||
train_kwarg["train_196"] = True
|
||||
train_kwarg["train_301"] = True
|
||||
elif intensity >= 9000: # NOT 9000!
|
||||
train_kwarg["train_dialogue"] = True
|
||||
train_kwarg["train_196"] = True
|
||||
train_kwarg["train_301"] = True
|
||||
|
||||
trainer = UbuntuCorpusTrainer2(self.chatbot, cog_data_path(self))
|
||||
return await trainer.asynctrain(**train_kwarg)
|
||||
|
||||
def _train_english(self):
|
||||
trainer = ChatterBotCorpusTrainer(self.chatbot)
|
||||
# try:
|
||||
@ -176,13 +213,10 @@ class Chatter(Cog):
|
||||
def _train(self, data):
|
||||
trainer = ListTrainer(self.chatbot)
|
||||
total = len(data)
|
||||
# try:
|
||||
for c, convo in enumerate(data, 1):
|
||||
log.info(f"{c} / {total}")
|
||||
if len(convo) > 1: # TODO: Toggleable skipping short conversations
|
||||
print(f"{c} / {total}")
|
||||
trainer.train(convo)
|
||||
# except:
|
||||
# return False
|
||||
return True
|
||||
|
||||
@commands.group(invoke_without_command=False)
|
||||
@ -190,10 +224,10 @@ class Chatter(Cog):
|
||||
"""
|
||||
Base command for this cog. Check help for the commands list.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
self._guild_cache[ctx.guild.id] = {} # Clear cache when modifying values
|
||||
self._global_cache = {}
|
||||
|
||||
@checks.admin()
|
||||
@commands.admin()
|
||||
@chatter.command(name="channel")
|
||||
async def chatter_channel(
|
||||
self, ctx: commands.Context, channel: Optional[discord.TextChannel] = None
|
||||
@ -213,13 +247,55 @@ class Chatter(Cog):
|
||||
await self.config.guild(ctx.guild).chatchannel.set(channel.id)
|
||||
await ctx.maybe_send_embed(f"Chat channel is now {channel.mention}")
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.admin()
|
||||
@chatter.command(name="reply")
|
||||
async def chatter_reply(self, ctx: commands.Context, toggle: Optional[bool] = None):
|
||||
"""
|
||||
Toggle bot reply to messages if conversation continuity is not present
|
||||
|
||||
"""
|
||||
reply = await self.config.guild(ctx.guild).reply()
|
||||
if toggle is None:
|
||||
toggle = not reply
|
||||
await self.config.guild(ctx.guild).reply.set(toggle)
|
||||
|
||||
if toggle:
|
||||
await ctx.maybe_send_embed(
|
||||
"I will now respond to you if conversation continuity is not present"
|
||||
)
|
||||
else:
|
||||
await ctx.maybe_send_embed(
|
||||
"I will not reply to your message if conversation continuity is not present, anymore"
|
||||
)
|
||||
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="learning")
|
||||
async def chatter_learning(self, ctx: commands.Context, toggle: Optional[bool] = None):
|
||||
"""
|
||||
Toggle the bot learning from its conversations.
|
||||
|
||||
This is a global setting.
|
||||
This is on by default.
|
||||
"""
|
||||
learning = await self.config.learning()
|
||||
if toggle is None:
|
||||
toggle = not learning
|
||||
await self.config.learning.set(toggle)
|
||||
|
||||
if toggle:
|
||||
await ctx.maybe_send_embed("I will now learn from conversations.")
|
||||
else:
|
||||
await ctx.maybe_send_embed("I will no longer learn from conversations.")
|
||||
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="cleardata")
|
||||
async def chatter_cleardata(self, ctx: commands.Context, confirm: bool = False):
|
||||
"""
|
||||
This command will erase all training data and reset your configuration settings
|
||||
This command will erase all training data and reset your configuration settings.
|
||||
|
||||
Use `[p]chatter cleardata True`
|
||||
This applies to all guilds.
|
||||
|
||||
Use `[p]chatter cleardata True` to confirm.
|
||||
"""
|
||||
|
||||
if not confirm:
|
||||
@ -246,7 +322,7 @@ class Chatter(Cog):
|
||||
|
||||
await ctx.tick()
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="algorithm", aliases=["algo"])
|
||||
async def chatter_algorithm(
|
||||
self, ctx: commands.Context, algo_number: int, threshold: float = None
|
||||
@ -280,7 +356,7 @@ class Chatter(Cog):
|
||||
|
||||
await ctx.tick()
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="model")
|
||||
async def chatter_model(self, ctx: commands.Context, model_number: int):
|
||||
"""
|
||||
@ -318,7 +394,7 @@ class Chatter(Cog):
|
||||
f"Model has been switched to {self.tagger_language.ISO_639_1}"
|
||||
)
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="minutes")
|
||||
async def minutes(self, ctx: commands.Context, minutes: int):
|
||||
"""
|
||||
@ -330,11 +406,11 @@ class Chatter(Cog):
|
||||
await ctx.send_help()
|
||||
return
|
||||
|
||||
await self.config.guild(ctx.guild).convo_length.set(minutes)
|
||||
await self.config.guild(ctx.guild).convo_delta.set(minutes)
|
||||
|
||||
await ctx.tick()
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="age")
|
||||
async def age(self, ctx: commands.Context, days: int):
|
||||
"""
|
||||
@ -349,7 +425,16 @@ class Chatter(Cog):
|
||||
await self.config.guild(ctx.guild).days.set(days)
|
||||
await ctx.tick()
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="kaggle")
|
||||
async def chatter_kaggle(self, ctx: commands.Context):
|
||||
"""Register with the kaggle API to download additional datasets for training"""
|
||||
if not await self.check_for_kaggle():
|
||||
await ctx.maybe_send_embed(
|
||||
"[Click here for instructions to setup the kaggle api](https://github.com/Kaggle/kaggle-api#api-credentials)"
|
||||
)
|
||||
|
||||
@commands.is_owner()
|
||||
@chatter.command(name="backup")
|
||||
async def backup(self, ctx, backupname):
|
||||
"""
|
||||
@ -371,8 +456,71 @@ class Chatter(Cog):
|
||||
else:
|
||||
await ctx.maybe_send_embed("Error occurred :(")
|
||||
|
||||
@checks.is_owner()
|
||||
@chatter.command(name="trainubuntu")
|
||||
@commands.is_owner()
|
||||
@chatter.group(name="train")
|
||||
async def chatter_train(self, ctx: commands.Context):
|
||||
"""Commands for training the bot"""
|
||||
pass
|
||||
|
||||
@chatter_train.group(name="kaggle")
|
||||
async def chatter_train_kaggle(self, ctx: commands.Context):
|
||||
"""
|
||||
Base command for kaggle training sets.
|
||||
|
||||
See `[p]chatter kaggle` for details on how to enable this option
|
||||
"""
|
||||
pass
|
||||
|
||||
@chatter_train_kaggle.command(name="ubuntu")
|
||||
async def chatter_train_kaggle_ubuntu(
|
||||
self, ctx: commands.Context, confirmation: bool = False, intensity=0
|
||||
):
|
||||
"""
|
||||
WARNING: Large Download! Trains the bot using *NEW* Ubuntu Dialog Corpus data.
|
||||
"""
|
||||
|
||||
if not confirmation:
|
||||
await ctx.maybe_send_embed(
|
||||
"Warning: This command downloads ~800MB and is CPU intensive during training\n"
|
||||
"If you're sure you want to continue, run `[p]chatter train kaggle ubuntu True`"
|
||||
)
|
||||
return
|
||||
|
||||
async with ctx.typing():
|
||||
future = await self._train_ubuntu2(intensity)
|
||||
|
||||
if future:
|
||||
await ctx.maybe_send_embed("Training successful!")
|
||||
else:
|
||||
await ctx.maybe_send_embed("Error occurred :(")
|
||||
|
||||
@chatter_train_kaggle.command(name="movies")
|
||||
async def chatter_train_kaggle_movies(self, ctx: commands.Context, confirmation: bool = False):
|
||||
"""
|
||||
WARNING: Language! Trains the bot using Cornell University's "Movie Dialog Corpus".
|
||||
|
||||
This training set contains dialog from a spread of movies with different MPAA.
|
||||
This dialog includes racism, sexism, and any number of sensitive topics.
|
||||
|
||||
Use at your own risk.
|
||||
"""
|
||||
|
||||
if not confirmation:
|
||||
await ctx.maybe_send_embed(
|
||||
"Warning: This command downloads ~29MB and is CPU intensive during training\n"
|
||||
"If you're sure you want to continue, run `[p]chatter train kaggle movies True`"
|
||||
)
|
||||
return
|
||||
|
||||
async with ctx.typing():
|
||||
future = await self._train_movies()
|
||||
|
||||
if future:
|
||||
await ctx.maybe_send_embed("Training successful!")
|
||||
else:
|
||||
await ctx.maybe_send_embed("Error occurred :(")
|
||||
|
||||
@chatter_train.command(name="ubuntu")
|
||||
async def chatter_train_ubuntu(self, ctx: commands.Context, confirmation: bool = False):
|
||||
"""
|
||||
WARNING: Large Download! Trains the bot using Ubuntu Dialog Corpus data.
|
||||
@ -380,8 +528,8 @@ class Chatter(Cog):
|
||||
|
||||
if not confirmation:
|
||||
await ctx.maybe_send_embed(
|
||||
"Warning: This command downloads ~500MB then eats your CPU for training\n"
|
||||
"If you're sure you want to continue, run `[p]chatter trainubuntu True`"
|
||||
"Warning: This command downloads ~500MB and is CPU intensive during training\n"
|
||||
"If you're sure you want to continue, run `[p]chatter train ubuntu True`"
|
||||
)
|
||||
return
|
||||
|
||||
@ -389,12 +537,11 @@ class Chatter(Cog):
|
||||
future = await self.loop.run_in_executor(None, self._train_ubuntu)
|
||||
|
||||
if future:
|
||||
await ctx.send("Training successful!")
|
||||
await ctx.maybe_send_embed("Training successful!")
|
||||
else:
|
||||
await ctx.send("Error occurred :(")
|
||||
await ctx.maybe_send_embed("Error occurred :(")
|
||||
|
||||
@checks.is_owner()
|
||||
@chatter.command(name="trainenglish")
|
||||
@chatter_train.command(name="english")
|
||||
async def chatter_train_english(self, ctx: commands.Context):
|
||||
"""
|
||||
Trains the bot in english
|
||||
@ -407,12 +554,32 @@ class Chatter(Cog):
|
||||
else:
|
||||
await ctx.maybe_send_embed("Error occurred :(")
|
||||
|
||||
@checks.is_owner()
|
||||
@chatter.command()
|
||||
async def train(self, ctx: commands.Context, channel: discord.TextChannel):
|
||||
@chatter_train.command(name="list")
|
||||
async def chatter_train_list(self, ctx: commands.Context):
|
||||
"""Trains the bot based on an uploaded list.
|
||||
|
||||
Must be a file in the format of a python list: ['prompt', 'response1', 'response2']
|
||||
"""
|
||||
Trains the bot based on language in this guild
|
||||
if not ctx.message.attachments:
|
||||
await ctx.maybe_send_embed("You must upload a file when using this command")
|
||||
return
|
||||
|
||||
attachment: discord.Attachment = ctx.message.attachments[0]
|
||||
|
||||
a_bytes = await attachment.read()
|
||||
|
||||
await ctx.send("Not yet implemented")
|
||||
|
||||
@chatter_train.command(name="channel")
|
||||
async def chatter_train_channel(
|
||||
self, ctx: commands.Context, channels: commands.Greedy[discord.TextChannel]
|
||||
):
|
||||
"""
|
||||
Trains the bot based on language in this guild.
|
||||
"""
|
||||
if not channels:
|
||||
await ctx.send_help()
|
||||
return
|
||||
|
||||
await ctx.maybe_send_embed(
|
||||
"Warning: The cog may use significant RAM or CPU if trained on large data sets.\n"
|
||||
@ -421,7 +588,7 @@ class Chatter(Cog):
|
||||
)
|
||||
|
||||
async with ctx.typing():
|
||||
conversation = await self._get_conversation(ctx, channel)
|
||||
conversation = await self._get_conversation(ctx, channels)
|
||||
|
||||
if not conversation:
|
||||
await ctx.maybe_send_embed("Failed to gather training data")
|
||||
@ -475,7 +642,18 @@ class Chatter(Cog):
|
||||
# Thank you Cog-Creators
|
||||
channel: discord.TextChannel = message.channel
|
||||
|
||||
if guild is not None and channel.id == await self.config.guild(guild).chatchannel():
|
||||
if not self._guild_cache[guild.id]:
|
||||
self._guild_cache[guild.id] = await self.config.guild(guild).all()
|
||||
|
||||
is_reply = False # this is only useful with in_response_to
|
||||
if (
|
||||
message.reference is not None
|
||||
and isinstance(message.reference.resolved, discord.Message)
|
||||
and message.reference.resolved.author.id == self.bot.user.id
|
||||
):
|
||||
is_reply = True # this is only useful with in_response_to
|
||||
pass # this is a reply to the bot, good to go
|
||||
elif guild is not None and channel.id == self._guild_cache[guild.id]["chatchannel"]:
|
||||
pass # good to go
|
||||
else:
|
||||
when_mentionables = commands.when_mentioned(self.bot, message)
|
||||
@ -490,10 +668,55 @@ class Chatter(Cog):
|
||||
|
||||
text = message.clean_content
|
||||
|
||||
async with channel.typing():
|
||||
future = await self.loop.run_in_executor(None, self.chatbot.get_response, text)
|
||||
async with ctx.typing():
|
||||
|
||||
if is_reply:
|
||||
in_response_to = message.reference.resolved.content
|
||||
elif self._last_message_per_channel[ctx.channel.id] is not None:
|
||||
last_m: discord.Message = self._last_message_per_channel[ctx.channel.id]
|
||||
minutes = self._guild_cache[ctx.guild.id]["convo_delta"]
|
||||
if (datetime.utcnow() - last_m.created_at).seconds > minutes * 60:
|
||||
in_response_to = None
|
||||
else:
|
||||
in_response_to = last_m.content
|
||||
else:
|
||||
in_response_to = None
|
||||
|
||||
# Always use generate reponse
|
||||
# Chatterbot tries to learn based on the result it comes up with, which is dumb
|
||||
log.debug("Generating response")
|
||||
Statement = self.chatbot.storage.get_object("statement")
|
||||
future = await self.loop.run_in_executor(
|
||||
None, self.chatbot.generate_response, Statement(text)
|
||||
)
|
||||
|
||||
if not self._global_cache:
|
||||
self._global_cache = await self.config.all()
|
||||
|
||||
if in_response_to is not None and self._global_cache["learning"]:
|
||||
log.debug("learning response")
|
||||
await self.loop.run_in_executor(
|
||||
None,
|
||||
partial(
|
||||
self.chatbot.learn_response,
|
||||
Statement(text),
|
||||
previous_statement=in_response_to,
|
||||
),
|
||||
)
|
||||
|
||||
replying = None
|
||||
if self._guild_cache[guild.id]["reply"]:
|
||||
if message != ctx.channel.last_message:
|
||||
replying = message
|
||||
|
||||
if future and str(future):
|
||||
await channel.send(str(future))
|
||||
self._last_message_per_channel[ctx.channel.id] = await channel.send(
|
||||
str(future), reference=replying
|
||||
)
|
||||
else:
|
||||
await channel.send(":thinking:")
|
||||
await ctx.send(":thinking:")
|
||||
|
||||
async def check_for_kaggle(self):
|
||||
"""Check whether Kaggle is installed and configured properly"""
|
||||
# TODO: This
|
||||
return False
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
"author": [
|
||||
"Bobloy"
|
||||
],
|
||||
"min_bot_version": "3.4.0",
|
||||
"min_bot_version": "3.4.6",
|
||||
"description": "Create an offline chatbot that talks like your average member using Machine Learning. See setup instructions at https://github.com/bobloy/Fox-V3/tree/master/chatter",
|
||||
"hidden": false,
|
||||
"install_msg": "Thank you for installing Chatter! Please make sure you check the install instructions at https://github.com/bobloy/Fox-V3/blob/master/chatter/README.md\nAfter that, get started ith `[p]load chatter` and `[p]help Chatter`",
|
||||
@ -17,7 +17,8 @@
|
||||
"pytz",
|
||||
"https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-2.3.1/en_core_web_sm-2.3.1.tar.gz#egg=en_core_web_sm",
|
||||
"https://github.com/explosion/spacy-models/releases/download/en_core_web_md-2.3.1/en_core_web_md-2.3.1.tar.gz#egg=en_core_web_md",
|
||||
"spacy>=2.3,<2.4"
|
||||
"spacy>=2.3,<2.4",
|
||||
"kaggle"
|
||||
],
|
||||
"short": "Local Chatbot run on machine learning",
|
||||
"end_user_data_statement": "This cog only stores anonymous conversations data; no End User Data is stored.",
|
||||
|
||||
73
chatter/storage_adapters.py
Normal file
73
chatter/storage_adapters.py
Normal file
@ -0,0 +1,73 @@
|
||||
from chatterbot.storage import StorageAdapter, SQLStorageAdapter
|
||||
|
||||
|
||||
class MyDumbSQLStorageAdapter(SQLStorageAdapter):
|
||||
def __init__(self, **kwargs):
|
||||
super(SQLStorageAdapter, self).__init__(**kwargs)
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
self.database_uri = kwargs.get("database_uri", False)
|
||||
|
||||
# None results in a sqlite in-memory database as the default
|
||||
if self.database_uri is None:
|
||||
self.database_uri = "sqlite://"
|
||||
|
||||
# Create a file database if the database is not a connection string
|
||||
if not self.database_uri:
|
||||
self.database_uri = "sqlite:///db.sqlite3"
|
||||
|
||||
self.engine = create_engine(
|
||||
self.database_uri, convert_unicode=True, connect_args={"check_same_thread": False}
|
||||
)
|
||||
|
||||
if self.database_uri.startswith("sqlite://"):
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy import event
|
||||
|
||||
@event.listens_for(Engine, "connect")
|
||||
def set_sqlite_pragma(dbapi_connection, connection_record):
|
||||
dbapi_connection.execute("PRAGMA journal_mode=WAL")
|
||||
dbapi_connection.execute("PRAGMA synchronous=NORMAL")
|
||||
|
||||
if not self.engine.dialect.has_table(self.engine, "Statement"):
|
||||
self.create_database()
|
||||
|
||||
self.Session = sessionmaker(bind=self.engine, expire_on_commit=True)
|
||||
|
||||
|
||||
class AsyncSQLStorageAdapter(SQLStorageAdapter):
|
||||
def __init__(self, **kwargs):
|
||||
super(SQLStorageAdapter, self).__init__(**kwargs)
|
||||
|
||||
self.database_uri = kwargs.get("database_uri", False)
|
||||
|
||||
# None results in a sqlite in-memory database as the default
|
||||
if self.database_uri is None:
|
||||
self.database_uri = "sqlite://"
|
||||
|
||||
# Create a file database if the database is not a connection string
|
||||
if not self.database_uri:
|
||||
self.database_uri = "sqlite:///db.sqlite3"
|
||||
|
||||
async def initialize(self):
|
||||
# from sqlalchemy import create_engine
|
||||
from aiomysql.sa import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
self.engine = await create_engine(self.database_uri, convert_unicode=True)
|
||||
|
||||
if self.database_uri.startswith("sqlite://"):
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy import event
|
||||
|
||||
@event.listens_for(Engine, "connect")
|
||||
def set_sqlite_pragma(dbapi_connection, connection_record):
|
||||
dbapi_connection.execute("PRAGMA journal_mode=WAL")
|
||||
dbapi_connection.execute("PRAGMA synchronous=NORMAL")
|
||||
|
||||
if not self.engine.dialect.has_table(self.engine, "Statement"):
|
||||
self.create_database()
|
||||
|
||||
self.Session = sessionmaker(bind=self.engine, expire_on_commit=True)
|
||||
351
chatter/trainers.py
Normal file
351
chatter/trainers.py
Normal file
@ -0,0 +1,351 @@
|
||||
import asyncio
|
||||
import csv
|
||||
import html
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
import time
|
||||
from functools import partial
|
||||
|
||||
from chatterbot import utils
|
||||
from chatterbot.conversation import Statement
|
||||
from chatterbot.tagging import PosLemmaTagger
|
||||
from chatterbot.trainers import Trainer
|
||||
from redbot.core.bot import Red
|
||||
from dateutil import parser as date_parser
|
||||
from redbot.core.utils import AsyncIter
|
||||
|
||||
log = logging.getLogger("red.fox_v3.chatter.trainers")
|
||||
|
||||
|
||||
class KaggleTrainer(Trainer):
|
||||
def __init__(self, chatbot, datapath: pathlib.Path, **kwargs):
|
||||
super().__init__(chatbot, **kwargs)
|
||||
|
||||
self.data_directory = datapath / kwargs.get("downloadpath", "kaggle_download")
|
||||
|
||||
self.kaggle_dataset = kwargs.get(
|
||||
"kaggle_dataset",
|
||||
"Cornell-University/movie-dialog-corpus",
|
||||
)
|
||||
|
||||
# Create the data directory if it does not already exist
|
||||
if not os.path.exists(self.data_directory):
|
||||
os.makedirs(self.data_directory)
|
||||
|
||||
def is_downloaded(self, file_path):
|
||||
"""
|
||||
Check if the data file is already downloaded.
|
||||
"""
|
||||
if os.path.exists(file_path):
|
||||
self.chatbot.logger.info("File is already downloaded")
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
async def download(self, dataset):
|
||||
import kaggle # This triggers the API token check
|
||||
|
||||
future = await asyncio.get_event_loop().run_in_executor(
|
||||
None,
|
||||
partial(
|
||||
kaggle.api.dataset_download_files,
|
||||
dataset=dataset,
|
||||
path=self.data_directory,
|
||||
quiet=False,
|
||||
unzip=True,
|
||||
),
|
||||
)
|
||||
|
||||
def train(self, *args, **kwargs):
|
||||
log.error("See asynctrain instead")
|
||||
|
||||
def asynctrain(self, *args, **kwargs):
|
||||
raise self.TrainerInitializationException()
|
||||
|
||||
|
||||
class SouthParkTrainer(KaggleTrainer):
|
||||
def __init__(self, chatbot, datapath: pathlib.Path, **kwargs):
|
||||
super().__init__(
|
||||
chatbot,
|
||||
datapath,
|
||||
downloadpath="ubuntu_data_v2",
|
||||
kaggle_dataset="tovarischsukhov/southparklines",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
class MovieTrainer(KaggleTrainer):
|
||||
def __init__(self, chatbot, datapath: pathlib.Path, **kwargs):
|
||||
super().__init__(
|
||||
chatbot,
|
||||
datapath,
|
||||
downloadpath="kaggle_movies",
|
||||
kaggle_dataset="Cornell-University/movie-dialog-corpus",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
async def run_movie_training(self):
|
||||
dialogue_file = "movie_lines.tsv"
|
||||
conversation_file = "movie_conversations.tsv"
|
||||
log.info(f"Beginning dialogue training on {dialogue_file}")
|
||||
start_time = time.time()
|
||||
|
||||
tagger = PosLemmaTagger(language=self.chatbot.storage.tagger.language)
|
||||
|
||||
# [lineID, characterID, movieID, character name, text of utterance]
|
||||
# File parsing from https://www.kaggle.com/mushaya/conversation-chatbot
|
||||
|
||||
with open(self.data_directory / conversation_file, "r", encoding="utf-8-sig") as conv_tsv:
|
||||
conv_lines = conv_tsv.readlines()
|
||||
with open(self.data_directory / dialogue_file, "r", encoding="utf-8-sig") as lines_tsv:
|
||||
dialog_lines = lines_tsv.readlines()
|
||||
|
||||
# trans_dict = str.maketrans({"<u>": "__", "</u>": "__", '""': '"'})
|
||||
|
||||
lines_dict = {}
|
||||
for line in dialog_lines:
|
||||
_line = line[:-1].strip('"').split("\t")
|
||||
if len(_line) >= 5: # Only good lines
|
||||
lines_dict[_line[0]] = (
|
||||
html.unescape(("".join(_line[4:])).strip())
|
||||
.replace("<u>", "__")
|
||||
.replace("</u>", "__")
|
||||
.replace('""', '"')
|
||||
)
|
||||
else:
|
||||
log.debug(f"Bad line {_line}")
|
||||
|
||||
# collecting line ids for each conversation
|
||||
conv = []
|
||||
for line in conv_lines[:-1]:
|
||||
_line = line[:-1].split("\t")[-1][1:-1].replace("'", "").replace(" ", ",")
|
||||
conv.append(_line.split(","))
|
||||
|
||||
# conversations = csv.reader(conv_tsv, delimiter="\t")
|
||||
#
|
||||
# reader = csv.reader(lines_tsv, delimiter="\t")
|
||||
#
|
||||
#
|
||||
#
|
||||
# lines_dict = {}
|
||||
# for row in reader:
|
||||
# try:
|
||||
# lines_dict[row[0].strip('"')] = row[4]
|
||||
# except:
|
||||
# log.exception(f"Bad line: {row}")
|
||||
# pass
|
||||
# else:
|
||||
# # log.info(f"Good line: {row}")
|
||||
# pass
|
||||
#
|
||||
# # lines_dict = {row[0].strip('"'): row[4] for row in reader_list}
|
||||
|
||||
statements_from_file = []
|
||||
save_every = 300
|
||||
count = 0
|
||||
|
||||
# [characterID of first, characterID of second, movieID, list of utterances]
|
||||
async for lines in AsyncIter(conv):
|
||||
previous_statement_text = None
|
||||
previous_statement_search_text = ""
|
||||
|
||||
for line in lines:
|
||||
text = lines_dict[line]
|
||||
statement = Statement(
|
||||
text=text,
|
||||
in_response_to=previous_statement_text,
|
||||
conversation="training",
|
||||
)
|
||||
|
||||
for preprocessor in self.chatbot.preprocessors:
|
||||
statement = preprocessor(statement)
|
||||
|
||||
statement.search_text = tagger.get_text_index_string(statement.text)
|
||||
statement.search_in_response_to = previous_statement_search_text
|
||||
|
||||
previous_statement_text = statement.text
|
||||
previous_statement_search_text = statement.search_text
|
||||
|
||||
statements_from_file.append(statement)
|
||||
|
||||
count += 1
|
||||
if count >= save_every:
|
||||
if statements_from_file:
|
||||
self.chatbot.storage.create_many(statements_from_file)
|
||||
statements_from_file = []
|
||||
count = 0
|
||||
|
||||
if statements_from_file:
|
||||
self.chatbot.storage.create_many(statements_from_file)
|
||||
|
||||
log.info(f"Training took {time.time() - start_time} seconds.")
|
||||
|
||||
async def asynctrain(self, *args, **kwargs):
|
||||
extracted_lines = self.data_directory / "movie_lines.tsv"
|
||||
extracted_lines: pathlib.Path
|
||||
|
||||
# Download and extract the Ubuntu dialog corpus if needed
|
||||
if not extracted_lines.exists():
|
||||
await self.download(self.kaggle_dataset)
|
||||
else:
|
||||
log.info("Movie dialog already downloaded")
|
||||
if not extracted_lines.exists():
|
||||
raise FileNotFoundError(f"{extracted_lines}")
|
||||
|
||||
await self.run_movie_training()
|
||||
|
||||
return True
|
||||
|
||||
# train_dialogue = kwargs.get("train_dialogue", True)
|
||||
# train_196_dialogue = kwargs.get("train_196", False)
|
||||
# train_301_dialogue = kwargs.get("train_301", False)
|
||||
#
|
||||
# if train_dialogue:
|
||||
# await self.run_dialogue_training(extracted_dir, "dialogueText.csv")
|
||||
#
|
||||
# if train_196_dialogue:
|
||||
# await self.run_dialogue_training(extracted_dir, "dialogueText_196.csv")
|
||||
#
|
||||
# if train_301_dialogue:
|
||||
# await self.run_dialogue_training(extracted_dir, "dialogueText_301.csv")
|
||||
|
||||
|
||||
class UbuntuCorpusTrainer2(KaggleTrainer):
|
||||
def __init__(self, chatbot, datapath: pathlib.Path, **kwargs):
|
||||
super().__init__(
|
||||
chatbot,
|
||||
datapath,
|
||||
downloadpath="kaggle_ubuntu",
|
||||
kaggle_dataset="rtatman/ubuntu-dialogue-corpus",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
async def asynctrain(self, *args, **kwargs):
|
||||
extracted_dir = self.data_directory / "Ubuntu-dialogue-corpus"
|
||||
|
||||
# Download and extract the Ubuntu dialog corpus if needed
|
||||
if not extracted_dir.exists():
|
||||
await self.download(self.kaggle_dataset)
|
||||
else:
|
||||
log.info("Ubuntu dialogue already downloaded")
|
||||
if not extracted_dir.exists():
|
||||
raise FileNotFoundError("Did not extract in the expected way")
|
||||
|
||||
train_dialogue = kwargs.get("train_dialogue", True)
|
||||
train_196_dialogue = kwargs.get("train_196", False)
|
||||
train_301_dialogue = kwargs.get("train_301", False)
|
||||
|
||||
if train_dialogue:
|
||||
await self.run_dialogue_training(extracted_dir, "dialogueText.csv")
|
||||
|
||||
if train_196_dialogue:
|
||||
await self.run_dialogue_training(extracted_dir, "dialogueText_196.csv")
|
||||
|
||||
if train_301_dialogue:
|
||||
await self.run_dialogue_training(extracted_dir, "dialogueText_301.csv")
|
||||
|
||||
return True
|
||||
|
||||
async def run_dialogue_training(self, extracted_dir, dialogue_file):
|
||||
log.info(f"Beginning dialogue training on {dialogue_file}")
|
||||
start_time = time.time()
|
||||
|
||||
tagger = PosLemmaTagger(language=self.chatbot.storage.tagger.language)
|
||||
|
||||
with open(extracted_dir / dialogue_file, "r", encoding="utf-8") as dg:
|
||||
reader = csv.DictReader(dg)
|
||||
|
||||
next(reader) # Skip the header
|
||||
|
||||
last_dialogue_id = None
|
||||
previous_statement_text = None
|
||||
previous_statement_search_text = ""
|
||||
statements_from_file = []
|
||||
|
||||
save_every = 50
|
||||
count = 0
|
||||
|
||||
async for row in AsyncIter(reader):
|
||||
dialogue_id = row["dialogueID"]
|
||||
if dialogue_id != last_dialogue_id:
|
||||
previous_statement_text = None
|
||||
previous_statement_search_text = ""
|
||||
last_dialogue_id = dialogue_id
|
||||
count += 1
|
||||
if count >= save_every:
|
||||
if statements_from_file:
|
||||
self.chatbot.storage.create_many(statements_from_file)
|
||||
statements_from_file = []
|
||||
count = 0
|
||||
|
||||
if len(row) > 0:
|
||||
statement = Statement(
|
||||
text=row["text"],
|
||||
in_response_to=previous_statement_text,
|
||||
conversation="training",
|
||||
# created_at=date_parser.parse(row["date"]),
|
||||
persona=row["from"],
|
||||
)
|
||||
|
||||
for preprocessor in self.chatbot.preprocessors:
|
||||
statement = preprocessor(statement)
|
||||
|
||||
statement.search_text = tagger.get_text_index_string(statement.text)
|
||||
statement.search_in_response_to = previous_statement_search_text
|
||||
|
||||
previous_statement_text = statement.text
|
||||
previous_statement_search_text = statement.search_text
|
||||
|
||||
statements_from_file.append(statement)
|
||||
|
||||
if statements_from_file:
|
||||
self.chatbot.storage.create_many(statements_from_file)
|
||||
|
||||
log.info(f"Training took {time.time() - start_time} seconds.")
|
||||
|
||||
|
||||
class TwitterCorpusTrainer(Trainer):
|
||||
pass
|
||||
# def train(self, *args, **kwargs):
|
||||
# """
|
||||
# Train the chat bot based on the provided list of
|
||||
# statements that represents a single conversation.
|
||||
# """
|
||||
# import twint
|
||||
#
|
||||
# c = twint.Config()
|
||||
# c.__dict__.update(kwargs)
|
||||
# twint.run.Search(c)
|
||||
#
|
||||
#
|
||||
# previous_statement_text = None
|
||||
# previous_statement_search_text = ''
|
||||
#
|
||||
# statements_to_create = []
|
||||
#
|
||||
# for conversation_count, text in enumerate(conversation):
|
||||
# if self.show_training_progress:
|
||||
# utils.print_progress_bar(
|
||||
# 'List Trainer',
|
||||
# conversation_count + 1, len(conversation)
|
||||
# )
|
||||
#
|
||||
# statement_search_text = self.chatbot.storage.tagger.get_text_index_string(text)
|
||||
#
|
||||
# statement = self.get_preprocessed_statement(
|
||||
# Statement(
|
||||
# text=text,
|
||||
# search_text=statement_search_text,
|
||||
# in_response_to=previous_statement_text,
|
||||
# search_in_response_to=previous_statement_search_text,
|
||||
# conversation='training'
|
||||
# )
|
||||
# )
|
||||
#
|
||||
# previous_statement_text = statement.text
|
||||
# previous_statement_search_text = statement_search_text
|
||||
#
|
||||
# statements_to_create.append(statement)
|
||||
#
|
||||
# self.chatbot.storage.create_many(statements_to_create)
|
||||
@ -58,11 +58,7 @@ class CogLint(Cog):
|
||||
|
||||
future = await self.bot.loop.run_in_executor(None, lint.py_run, path, "return_std=True")
|
||||
|
||||
if future:
|
||||
(pylint_stdout, pylint_stderr) = future
|
||||
else:
|
||||
(pylint_stdout, pylint_stderr) = None, None
|
||||
|
||||
(pylint_stdout, pylint_stderr) = future or (None, None)
|
||||
# print(pylint_stderr)
|
||||
# print(pylint_stdout)
|
||||
|
||||
|
||||
@ -67,9 +67,8 @@ class Conquest(commands.Cog):
|
||||
"""
|
||||
Base command for conquest cog. Start with `[p]conquest set map` to select a map.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
if self.current_map is not None:
|
||||
await self._conquest_current(ctx)
|
||||
if ctx.invoked_subcommand is None and self.current_map is not None:
|
||||
await self._conquest_current(ctx)
|
||||
|
||||
@conquest.command(name="list")
|
||||
async def _conquest_list(self, ctx: commands.Context):
|
||||
@ -80,14 +79,13 @@ class Conquest(commands.Cog):
|
||||
|
||||
with maps_json.open() as maps:
|
||||
maps_json = json.load(maps)
|
||||
map_list = "\n".join(map_name for map_name in maps_json["maps"])
|
||||
map_list = "\n".join(maps_json["maps"])
|
||||
await ctx.maybe_send_embed(f"Current maps:\n{map_list}")
|
||||
|
||||
@conquest.group(name="set")
|
||||
async def conquest_set(self, ctx: commands.Context):
|
||||
"""Base command for admin actions like selecting a map"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@conquest_set.command(name="resetzoom")
|
||||
async def _conquest_set_resetzoom(self, ctx: commands.Context):
|
||||
|
||||
@ -30,8 +30,7 @@ class MapMaker(commands.Cog):
|
||||
"""
|
||||
Base command for managing current maps or creating new ones
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@mapmaker.command(name="upload")
|
||||
async def _mapmaker_upload(self, ctx: commands.Context, map_path=""):
|
||||
|
||||
@ -65,7 +65,7 @@ def floodfill(image, xy, value, border=None, thresh=0) -> set:
|
||||
if border is None:
|
||||
fill = _color_diff(p, background) <= thresh
|
||||
else:
|
||||
fill = p != value and p != border
|
||||
fill = p not in [value, border]
|
||||
if fill:
|
||||
pixel[s, t] = value
|
||||
new_edge.add((s, t))
|
||||
|
||||
@ -27,8 +27,7 @@ class ExclusiveRole(Cog):
|
||||
async def exclusive(self, ctx):
|
||||
"""Base command for managing exclusive roles"""
|
||||
|
||||
if not ctx.invoked_subcommand:
|
||||
pass
|
||||
pass
|
||||
|
||||
@exclusive.command(name="add")
|
||||
@checks.mod_or_permissions(administrator=True)
|
||||
@ -85,7 +84,7 @@ class ExclusiveRole(Cog):
|
||||
if role_set is None:
|
||||
role_set = set(await self.config.guild(member.guild).role_list())
|
||||
|
||||
member_set = set([role.id for role in member.roles])
|
||||
member_set = {role.id for role in member.roles}
|
||||
to_remove = (member_set - role_set) - {member.guild.default_role.id}
|
||||
|
||||
if to_remove and member_set & role_set:
|
||||
@ -103,7 +102,7 @@ class ExclusiveRole(Cog):
|
||||
await asyncio.sleep(1)
|
||||
|
||||
role_set = set(await self.config.guild(after.guild).role_list())
|
||||
member_set = set([role.id for role in after.roles])
|
||||
member_set = {role.id for role in after.roles}
|
||||
|
||||
if role_set & member_set:
|
||||
try:
|
||||
|
||||
15
fifo/fifo.py
15
fifo/fifo.py
@ -68,10 +68,7 @@ class CapturePrint:
|
||||
self.string = None
|
||||
|
||||
def write(self, string):
|
||||
if self.string is None:
|
||||
self.string = string
|
||||
else:
|
||||
self.string = self.string + "\n" + string
|
||||
self.string = string if self.string is None else self.string + "\n" + string
|
||||
|
||||
|
||||
class FIFO(commands.Cog):
|
||||
@ -197,8 +194,8 @@ class FIFO(commands.Cog):
|
||||
async def _get_tz(self, user: Union[discord.User, discord.Member]) -> Union[None, tzinfo]:
|
||||
if self.tz_cog is None:
|
||||
self.tz_cog = self.bot.get_cog("Timezone")
|
||||
if self.tz_cog is None:
|
||||
self.tz_cog = False # only try once to get the timezone cog
|
||||
if self.tz_cog is None:
|
||||
self.tz_cog = False # only try once to get the timezone cog
|
||||
|
||||
if not self.tz_cog:
|
||||
return None
|
||||
@ -230,8 +227,7 @@ class FIFO(commands.Cog):
|
||||
"""
|
||||
Base command for handling scheduling of tasks
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@fifo.command(name="wakeup")
|
||||
async def fifo_wakeup(self, ctx: commands.Context):
|
||||
@ -522,8 +518,7 @@ class FIFO(commands.Cog):
|
||||
"""
|
||||
Add a new trigger for a task from the current guild.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@fifo_trigger.command(name="interval")
|
||||
async def fifo_trigger_interval(
|
||||
|
||||
@ -53,12 +53,9 @@ class Flag(Cog):
|
||||
@commands.group()
|
||||
async def flagset(self, ctx: commands.Context):
|
||||
"""
|
||||
My custom cog
|
||||
|
||||
Extra information goes here
|
||||
Commands for managing Flag settings
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@flagset.command(name="expire")
|
||||
async def flagset_expire(self, ctx: commands.Context, days: int):
|
||||
|
||||
@ -147,8 +147,7 @@ class Hangman(Cog):
|
||||
@checks.mod_or_permissions(administrator=True)
|
||||
async def hangset(self, ctx):
|
||||
"""Adjust hangman settings"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@hangset.command()
|
||||
async def face(self, ctx: commands.Context, theface):
|
||||
@ -250,7 +249,7 @@ class Hangman(Cog):
|
||||
|
||||
self.winbool[guild] = True
|
||||
for i in self.the_data[guild]["answer"]:
|
||||
if i == " " or i == "-":
|
||||
if i in [" ", "-"]:
|
||||
out_str += i * 2
|
||||
elif i in self.the_data[guild]["guesses"] or i not in "ABCDEFGHIJKLMNOPQRSTUVWXYZ":
|
||||
out_str += "__" + i + "__ "
|
||||
@ -262,9 +261,7 @@ class Hangman(Cog):
|
||||
|
||||
def _guesslist(self, guild):
|
||||
"""Returns the current letter list"""
|
||||
out_str = ""
|
||||
for i in self.the_data[guild]["guesses"]:
|
||||
out_str += str(i) + ","
|
||||
out_str = "".join(str(i) + "," for i in self.the_data[guild]["guesses"])
|
||||
out_str = out_str[:-1]
|
||||
|
||||
return out_str
|
||||
|
||||
@ -65,9 +65,9 @@ class InfoChannel(Cog):
|
||||
"offline": "Offline: {count}",
|
||||
}
|
||||
|
||||
default_channel_ids = {k: None for k in self.default_channel_names.keys()}
|
||||
default_channel_ids = {k: None for k in self.default_channel_names}
|
||||
# Only members is enabled by default
|
||||
default_enabled_counts = {k: k == "members" for k in self.default_channel_names.keys()}
|
||||
default_enabled_counts = {k: k == "members" for k in self.default_channel_names}
|
||||
|
||||
default_guild = {
|
||||
"category_id": None,
|
||||
@ -159,8 +159,7 @@ class InfoChannel(Cog):
|
||||
"""
|
||||
Toggle different types of infochannels
|
||||
"""
|
||||
if not ctx.invoked_subcommand:
|
||||
pass
|
||||
pass
|
||||
|
||||
@infochannelset.command(name="togglechannel")
|
||||
async def _infochannelset_togglechannel(
|
||||
|
||||
@ -10,9 +10,9 @@ log = logging.getLogger("red.fox_v3.isitdown")
|
||||
|
||||
class IsItDown(commands.Cog):
|
||||
"""
|
||||
Cog Description
|
||||
Cog for checking whether a website is down or not.
|
||||
|
||||
Less important information about the cog
|
||||
Uses the `isitdown.site` API
|
||||
"""
|
||||
|
||||
def __init__(self, bot: Red):
|
||||
@ -36,23 +36,25 @@ class IsItDown(commands.Cog):
|
||||
Alias: iid
|
||||
"""
|
||||
try:
|
||||
resp = await self._check_if_down(url_to_check)
|
||||
resp, url = await self._check_if_down(url_to_check)
|
||||
except AssertionError:
|
||||
await ctx.maybe_send_embed("Invalid URL provided. Make sure not to include `http://`")
|
||||
return
|
||||
|
||||
# log.debug(resp)
|
||||
if resp["isitdown"]:
|
||||
await ctx.maybe_send_embed(f"{url_to_check} is DOWN!")
|
||||
await ctx.maybe_send_embed(f"{url} is DOWN!")
|
||||
else:
|
||||
await ctx.maybe_send_embed(f"{url_to_check} is UP!")
|
||||
await ctx.maybe_send_embed(f"{url} is UP!")
|
||||
|
||||
async def _check_if_down(self, url_to_check):
|
||||
url = re.compile(r"https?://(www\.)?")
|
||||
url.sub("", url_to_check).strip().strip("/")
|
||||
re_compiled = re.compile(r"https?://(www\.)?")
|
||||
url = re_compiled.sub("", url_to_check).strip().strip("/")
|
||||
|
||||
url = f"https://isitdown.site/api/v3/{url}"
|
||||
# log.debug(url)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as response:
|
||||
assert response.status == 200
|
||||
resp = await response.json()
|
||||
return resp
|
||||
return resp, url
|
||||
|
||||
@ -8,7 +8,7 @@
|
||||
"install_msg": "Thank you for installing LaunchLib. Get started with `[p]load launchlib`, then `[p]help LaunchLib`",
|
||||
"short": "Access launch data for space flights",
|
||||
"end_user_data_statement": "This cog does not store any End User Data",
|
||||
"requirements": ["python-launch-library>=1.0.6"],
|
||||
"requirements": ["python-launch-library>=2.0.3"],
|
||||
"tags": [
|
||||
"bobloy",
|
||||
"utils",
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
import asyncio
|
||||
import functools
|
||||
import logging
|
||||
|
||||
import re
|
||||
import discord
|
||||
import launchlibrary as ll
|
||||
from redbot.core import Config, commands
|
||||
@ -14,9 +14,7 @@ log = logging.getLogger("red.fox_v3.launchlib")
|
||||
|
||||
class LaunchLib(commands.Cog):
|
||||
"""
|
||||
Cog Description
|
||||
|
||||
Less important information about the cog
|
||||
Cog using `thespacedevs` API to get details about rocket launches
|
||||
"""
|
||||
|
||||
def __init__(self, bot: Red):
|
||||
@ -37,27 +35,30 @@ class LaunchLib(commands.Cog):
|
||||
return
|
||||
|
||||
async def _embed_launch_data(self, launch: ll.AsyncLaunch):
|
||||
status: ll.AsyncLaunchStatus = await launch.get_status()
|
||||
|
||||
# status: ll.AsyncLaunchStatus = await launch.get_status()
|
||||
status = launch.status
|
||||
|
||||
rocket: ll.AsyncRocket = launch.rocket
|
||||
|
||||
title = launch.name
|
||||
description = status.description
|
||||
description = status["name"]
|
||||
|
||||
urls = launch.vid_urls + launch.info_urls
|
||||
if not urls and rocket:
|
||||
urls = rocket.info_urls + [rocket.wiki_url]
|
||||
if urls:
|
||||
url = urls[0]
|
||||
else:
|
||||
url = None
|
||||
if rocket:
|
||||
urls += [rocket.info_url, rocket.wiki_url]
|
||||
if launch.pad:
|
||||
urls += [launch.pad.info_url, launch.pad.wiki_url]
|
||||
|
||||
color = discord.Color.green() if status.id in [1, 3] else discord.Color.red()
|
||||
url = next((url for url in urls if urls is not None), None) if urls else None
|
||||
color = discord.Color.green() if status["id"] in [1, 3] else discord.Color.red()
|
||||
|
||||
em = discord.Embed(title=title, description=description, url=url, color=color)
|
||||
|
||||
if rocket and rocket.image_url and rocket.image_url != "Array":
|
||||
em.set_image(url=rocket.image_url)
|
||||
elif launch.pad and launch.pad.map_image:
|
||||
em.set_image(url=launch.pad.map_image)
|
||||
|
||||
agency = getattr(launch, "agency", None)
|
||||
if agency is not None:
|
||||
@ -89,6 +90,18 @@ class LaunchLib(commands.Cog):
|
||||
data = mission.get(f[0], None)
|
||||
if data is not None and data:
|
||||
em.add_field(name=f[1], value=data)
|
||||
if launch.pad:
|
||||
location_url = getattr(launch.pad, "map_url", None)
|
||||
pad_name = getattr(launch.pad, "name", None)
|
||||
|
||||
if pad_name is not None:
|
||||
if location_url is not None:
|
||||
location_url = re.sub(
|
||||
"[^a-zA-Z0-9/:.'+\"°?=,-]", "", location_url
|
||||
) # Fix bad URLS
|
||||
em.add_field(name="Launch Pad Name", value=f"[{pad_name}]({location_url})")
|
||||
else:
|
||||
em.add_field(name="Launch Pad Name", value=pad_name)
|
||||
|
||||
if rocket and rocket.family:
|
||||
em.add_field(name="Rocket Family", value=rocket.family)
|
||||
@ -101,11 +114,16 @@ class LaunchLib(commands.Cog):
|
||||
|
||||
@commands.group()
|
||||
async def launchlib(self, ctx: commands.Context):
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
"""Base command for getting launches"""
|
||||
pass
|
||||
|
||||
@launchlib.command()
|
||||
async def next(self, ctx: commands.Context, num_launches: int = 1):
|
||||
"""
|
||||
Show the next launches
|
||||
|
||||
Use `num_launches` to get more than one.
|
||||
"""
|
||||
# launches = await api.async_next_launches(num_launches)
|
||||
# loop = asyncio.get_running_loop()
|
||||
#
|
||||
@ -115,6 +133,8 @@ class LaunchLib(commands.Cog):
|
||||
#
|
||||
launches = await self.api.async_fetch_launch(num=num_launches)
|
||||
|
||||
# log.debug(str(launches))
|
||||
|
||||
async with ctx.typing():
|
||||
for x, launch in enumerate(launches):
|
||||
if x >= num_launches:
|
||||
|
||||
@ -25,8 +25,7 @@ class Leaver(Cog):
|
||||
@checks.mod_or_permissions(administrator=True)
|
||||
async def leaverset(self, ctx):
|
||||
"""Adjust leaver settings"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@leaverset.command()
|
||||
async def channel(self, ctx: Context):
|
||||
@ -57,5 +56,3 @@ class Leaver(Cog):
|
||||
)
|
||||
else:
|
||||
await channel.send(out)
|
||||
else:
|
||||
pass
|
||||
|
||||
@ -45,14 +45,12 @@ class LastSeen(Cog):
|
||||
|
||||
@staticmethod
|
||||
def get_date_time(s):
|
||||
d = dateutil.parser.parse(s)
|
||||
return d
|
||||
return dateutil.parser.parse(s)
|
||||
|
||||
@commands.group(aliases=["setlseen"], name="lseenset")
|
||||
async def lset(self, ctx: commands.Context):
|
||||
"""Change settings for lseen"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@lset.command(name="toggle")
|
||||
async def lset_toggle(self, ctx: commands.Context):
|
||||
|
||||
@ -111,9 +111,8 @@ async def _withdraw_points(gardener: Gardener, amount):
|
||||
|
||||
if (gardener.points - amount) < 0:
|
||||
return False
|
||||
else:
|
||||
gardener.points -= amount
|
||||
return True
|
||||
gardener.points -= amount
|
||||
return True
|
||||
|
||||
|
||||
class PlantTycoon(commands.Cog):
|
||||
@ -245,11 +244,9 @@ class PlantTycoon(commands.Cog):
|
||||
await self._load_plants_products()
|
||||
|
||||
modifiers = sum(
|
||||
[
|
||||
self.products[product]["modifier"]
|
||||
for product in gardener.products
|
||||
if gardener.products[product] > 0
|
||||
]
|
||||
self.products[product]["modifier"]
|
||||
for product in gardener.products
|
||||
if gardener.products[product] > 0
|
||||
)
|
||||
|
||||
degradation = (
|
||||
@ -290,38 +287,31 @@ class PlantTycoon(commands.Cog):
|
||||
product = product.lower()
|
||||
product_category = product_category.lower()
|
||||
if product in self.products and self.products[product]["category"] == product_category:
|
||||
if product in gardener.products:
|
||||
if gardener.products[product] > 0:
|
||||
gardener.current["health"] += self.products[product]["health"]
|
||||
gardener.products[product] -= 1
|
||||
if gardener.products[product] == 0:
|
||||
del gardener.products[product.lower()]
|
||||
if product_category == "water":
|
||||
emoji = ":sweat_drops:"
|
||||
elif product_category == "fertilizer":
|
||||
emoji = ":poop:"
|
||||
# elif product_category == "tool":
|
||||
if product in gardener.products and gardener.products[product] > 0:
|
||||
gardener.current["health"] += self.products[product]["health"]
|
||||
gardener.products[product] -= 1
|
||||
if gardener.products[product] == 0:
|
||||
del gardener.products[product.lower()]
|
||||
if product_category == "fertilizer":
|
||||
emoji = ":poop:"
|
||||
elif product_category == "water":
|
||||
emoji = ":sweat_drops:"
|
||||
else:
|
||||
emoji = ":scissors:"
|
||||
message = "Your plant got some health back! {}".format(emoji)
|
||||
if gardener.current["health"] > gardener.current["threshold"]:
|
||||
gardener.current["health"] -= self.products[product]["damage"]
|
||||
if product_category == "tool":
|
||||
damage_msg = "You used {} too many times!".format(product)
|
||||
else:
|
||||
emoji = ":scissors:"
|
||||
message = "Your plant got some health back! {}".format(emoji)
|
||||
if gardener.current["health"] > gardener.current["threshold"]:
|
||||
gardener.current["health"] -= self.products[product]["damage"]
|
||||
if product_category == "tool":
|
||||
damage_msg = "You used {} too many times!".format(product)
|
||||
else:
|
||||
damage_msg = "You gave too much of {}.".format(product)
|
||||
message = "{} Your plant lost some health. :wilted_rose:".format(
|
||||
damage_msg
|
||||
)
|
||||
gardener.points += self.defaults["points"]["add_health"]
|
||||
await gardener.save_gardener()
|
||||
else:
|
||||
message = "You have no {}. Go buy some!".format(product)
|
||||
damage_msg = "You gave too much of {}.".format(product)
|
||||
message = "{} Your plant lost some health. :wilted_rose:".format(damage_msg)
|
||||
gardener.points += self.defaults["points"]["add_health"]
|
||||
await gardener.save_gardener()
|
||||
elif product in gardener.products or product_category != "tool":
|
||||
message = "You have no {}. Go buy some!".format(product)
|
||||
else:
|
||||
if product_category == "tool":
|
||||
message = "You don't have a {}. Go buy one!".format(product)
|
||||
else:
|
||||
message = "You have no {}. Go buy some!".format(product)
|
||||
message = "You don't have a {}. Go buy one!".format(product)
|
||||
else:
|
||||
message = "Are you sure you are using {}?".format(product_category)
|
||||
|
||||
@ -412,24 +402,18 @@ class PlantTycoon(commands.Cog):
|
||||
gardener.current = plant
|
||||
await gardener.save_gardener()
|
||||
|
||||
em = discord.Embed(description=message, color=discord.Color.green())
|
||||
else:
|
||||
plant = gardener.current
|
||||
message = "You're already growing {} **{}**, silly.".format(
|
||||
plant["article"], plant["name"]
|
||||
)
|
||||
em = discord.Embed(description=message, color=discord.Color.green())
|
||||
|
||||
em = discord.Embed(description=message, color=discord.Color.green())
|
||||
await ctx.send(embed=em)
|
||||
|
||||
@_gardening.command(name="profile")
|
||||
async def _profile(self, ctx: commands.Context, *, member: discord.Member = None):
|
||||
"""Check your gardening profile."""
|
||||
if member is not None:
|
||||
author = member
|
||||
else:
|
||||
author = ctx.author
|
||||
|
||||
author = member if member is not None else ctx.author
|
||||
gardener = await self._gardener(author)
|
||||
try:
|
||||
await self._apply_degradation(gardener)
|
||||
@ -440,9 +424,7 @@ class PlantTycoon(commands.Cog):
|
||||
avatar = author.avatar_url if author.avatar else author.default_avatar_url
|
||||
em.set_author(name="Gardening profile of {}".format(author.name), icon_url=avatar)
|
||||
em.add_field(name="**Thneeds**", value=str(gardener.points))
|
||||
if not gardener.current:
|
||||
em.add_field(name="**Currently growing**", value="None")
|
||||
else:
|
||||
if gardener.current:
|
||||
em.set_thumbnail(url=gardener.current["image"])
|
||||
em.add_field(
|
||||
name="**Currently growing**",
|
||||
@ -450,16 +432,15 @@ class PlantTycoon(commands.Cog):
|
||||
gardener.current["name"], gardener.current["health"]
|
||||
),
|
||||
)
|
||||
else:
|
||||
em.add_field(name="**Currently growing**", value="None")
|
||||
if not gardener.badges:
|
||||
em.add_field(name="**Badges**", value="None")
|
||||
else:
|
||||
badges = ""
|
||||
for badge in gardener.badges:
|
||||
badges += "{}\n".format(badge.capitalize())
|
||||
badges = "".join("{}\n".format(badge.capitalize()) for badge in gardener.badges)
|
||||
|
||||
em.add_field(name="**Badges**", value=badges)
|
||||
if not gardener.products:
|
||||
em.add_field(name="**Products**", value="None")
|
||||
else:
|
||||
if gardener.products:
|
||||
products = ""
|
||||
for product_name, product_data in gardener.products.items():
|
||||
if self.products[product_name] is None:
|
||||
@ -470,6 +451,8 @@ class PlantTycoon(commands.Cog):
|
||||
self.products[product_name]["modifier"],
|
||||
)
|
||||
em.add_field(name="**Products**", value=products)
|
||||
else:
|
||||
em.add_field(name="**Products**", value="None")
|
||||
if gardener.current:
|
||||
degradation = await self._degradation(gardener)
|
||||
die_in = await _die_in(gardener, degradation)
|
||||
@ -600,7 +583,6 @@ class PlantTycoon(commands.Cog):
|
||||
self.products[pd]["category"],
|
||||
),
|
||||
)
|
||||
await ctx.send(embed=em)
|
||||
else:
|
||||
if amount <= 0:
|
||||
message = "Invalid amount! Must be greater than 1"
|
||||
@ -629,7 +611,8 @@ class PlantTycoon(commands.Cog):
|
||||
else:
|
||||
message = "I don't have this product."
|
||||
em = discord.Embed(description=message, color=discord.Color.green())
|
||||
await ctx.send(embed=em)
|
||||
|
||||
await ctx.send(embed=em)
|
||||
|
||||
@_gardening.command(name="convert")
|
||||
async def _convert(self, ctx: commands.Context, amount: int):
|
||||
@ -663,8 +646,7 @@ class PlantTycoon(commands.Cog):
|
||||
else:
|
||||
gardener.current = {}
|
||||
message = "You successfully shovelled your plant out."
|
||||
if gardener.points < 0:
|
||||
gardener.points = 0
|
||||
gardener.points = max(gardener.points, 0)
|
||||
await gardener.save_gardener()
|
||||
|
||||
em = discord.Embed(description=message, color=discord.Color.dark_grey())
|
||||
@ -681,12 +663,12 @@ class PlantTycoon(commands.Cog):
|
||||
except discord.Forbidden:
|
||||
# Couldn't DM the degradation
|
||||
await ctx.send("ERROR\nYou blocked me, didn't you?")
|
||||
product = "water"
|
||||
product_category = "water"
|
||||
if not gardener.current:
|
||||
message = "You're currently not growing a plant."
|
||||
await _send_message(channel, message)
|
||||
else:
|
||||
product = "water"
|
||||
product_category = "water"
|
||||
await self._add_health(channel, gardener, product, product_category)
|
||||
|
||||
@commands.command(name="fertilize")
|
||||
@ -700,11 +682,11 @@ class PlantTycoon(commands.Cog):
|
||||
await ctx.send("ERROR\nYou blocked me, didn't you?")
|
||||
channel = ctx.channel
|
||||
product = fertilizer
|
||||
product_category = "fertilizer"
|
||||
if not gardener.current:
|
||||
message = "You're currently not growing a plant."
|
||||
await _send_message(channel, message)
|
||||
else:
|
||||
product_category = "fertilizer"
|
||||
await self._add_health(channel, gardener, product, product_category)
|
||||
|
||||
@commands.command(name="prune")
|
||||
@ -717,12 +699,12 @@ class PlantTycoon(commands.Cog):
|
||||
# Couldn't DM the degradation
|
||||
await ctx.send("ERROR\nYou blocked me, didn't you?")
|
||||
channel = ctx.channel
|
||||
product = "pruner"
|
||||
product_category = "tool"
|
||||
if not gardener.current:
|
||||
message = "You're currently not growing a plant."
|
||||
await _send_message(channel, message)
|
||||
else:
|
||||
product = "pruner"
|
||||
product_category = "tool"
|
||||
await self._add_health(channel, gardener, product, product_category)
|
||||
|
||||
# async def check_degradation(self):
|
||||
|
||||
@ -67,8 +67,10 @@ class QRInvite(Cog):
|
||||
|
||||
extension = pathlib.Path(image_url).parts[-1].replace(".", "?").split("?")[1]
|
||||
|
||||
save_as_name = f"{ctx.guild.id}-{ctx.author.id}"
|
||||
|
||||
path: pathlib.Path = cog_data_path(self)
|
||||
image_path = path / (ctx.guild.icon + "." + extension)
|
||||
image_path = path / f"{save_as_name}.{extension}"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(image_url) as response:
|
||||
image = await response.read()
|
||||
@ -77,27 +79,29 @@ class QRInvite(Cog):
|
||||
file.write(image)
|
||||
|
||||
if extension == "webp":
|
||||
new_path = convert_webp_to_png(str(image_path))
|
||||
new_image_path = convert_webp_to_png(str(image_path))
|
||||
elif extension == "gif":
|
||||
await ctx.maybe_send_embed("gif is not supported yet, stay tuned")
|
||||
return
|
||||
elif extension == "png":
|
||||
new_path = str(image_path)
|
||||
new_image_path = str(image_path)
|
||||
elif extension == "jpg":
|
||||
new_image_path = convert_jpg_to_png(str(image_path))
|
||||
else:
|
||||
await ctx.maybe_send_embed(f"{extension} is not supported yet, stay tuned")
|
||||
return
|
||||
|
||||
myqr.run(
|
||||
invite,
|
||||
picture=new_path,
|
||||
save_name=ctx.guild.icon + "_qrcode.png",
|
||||
picture=new_image_path,
|
||||
save_name=f"{save_as_name}_qrcode.png",
|
||||
save_dir=str(cog_data_path(self)),
|
||||
colorized=colorized,
|
||||
)
|
||||
|
||||
png_path: pathlib.Path = path / (ctx.guild.icon + "_qrcode.png")
|
||||
with png_path.open("rb") as png_fp:
|
||||
await ctx.send(file=discord.File(png_fp.read(), "qrcode.png"))
|
||||
png_path: pathlib.Path = path / f"{save_as_name}_qrcode.png"
|
||||
# with png_path.open("rb") as png_fp:
|
||||
await ctx.send(file=discord.File(png_path, "qrcode.png"))
|
||||
|
||||
|
||||
def convert_webp_to_png(path):
|
||||
@ -110,3 +114,10 @@ def convert_webp_to_png(path):
|
||||
new_path = path.replace(".webp", ".png")
|
||||
im.save(new_path, transparency=255)
|
||||
return new_path
|
||||
|
||||
|
||||
def convert_jpg_to_png(path):
|
||||
im = Image.open(path)
|
||||
new_path = path.replace(".jpg", ".png")
|
||||
im.save(new_path)
|
||||
return new_path
|
||||
|
||||
@ -97,9 +97,7 @@ class ReactRestrict(Cog):
|
||||
"""
|
||||
current_combos = await self.combo_list()
|
||||
|
||||
to_keep = [
|
||||
c for c in current_combos if not (c.message_id == message_id and c.role_id == role.id)
|
||||
]
|
||||
to_keep = [c for c in current_combos if c.message_id != message_id or c.role_id != role.id]
|
||||
|
||||
if to_keep != current_combos:
|
||||
await self.set_combo_list(to_keep)
|
||||
@ -210,8 +208,7 @@ class ReactRestrict(Cog):
|
||||
"""
|
||||
Base command for this cog. Check help for the commands list.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@reactrestrict.command()
|
||||
async def add(self, ctx: commands.Context, message_id: int, *, role: discord.Role):
|
||||
|
||||
@ -69,13 +69,12 @@ class RPSLS(Cog):
|
||||
|
||||
def get_emote(self, choice):
|
||||
if choice == "rock":
|
||||
emote = ":moyai:"
|
||||
return ":moyai:"
|
||||
elif choice == "spock":
|
||||
emote = ":vulcan:"
|
||||
return ":vulcan:"
|
||||
elif choice == "paper":
|
||||
emote = ":page_facing_up:"
|
||||
return ":page_facing_up:"
|
||||
elif choice in ["scissors", "lizard"]:
|
||||
emote = ":{}:".format(choice)
|
||||
return ":{}:".format(choice)
|
||||
else:
|
||||
emote = None
|
||||
return emote
|
||||
return None
|
||||
|
||||
@ -16,16 +16,16 @@ log = logging.getLogger("red.fox_v3.stealemoji")
|
||||
|
||||
|
||||
async def check_guild(guild, emoji):
|
||||
if len(guild.emojis) >= 100:
|
||||
if len(guild.emojis) >= 2 * guild.emoji_limit:
|
||||
return False
|
||||
|
||||
if len(guild.emojis) < 50:
|
||||
if len(guild.emojis) < guild.emoji_limit:
|
||||
return True
|
||||
|
||||
if emoji.animated:
|
||||
return sum(e.animated for e in guild.emojis) < 50
|
||||
return sum(e.animated for e in guild.emojis) < guild.emoji_limit
|
||||
else:
|
||||
return sum(not e.animated for e in guild.emojis) < 50
|
||||
return sum(not e.animated for e in guild.emojis) < guild.emoji_limit
|
||||
|
||||
|
||||
class StealEmoji(Cog):
|
||||
@ -69,8 +69,7 @@ class StealEmoji(Cog):
|
||||
"""
|
||||
Base command for this cog. Check help for the commands list.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@checks.is_owner()
|
||||
@stealemoji.command(name="clearemojis")
|
||||
@ -268,37 +267,36 @@ class StealEmoji(Cog):
|
||||
break
|
||||
|
||||
if guildbank is None:
|
||||
if await self.config.autobank():
|
||||
try:
|
||||
guildbank: discord.Guild = await self.bot.create_guild(
|
||||
"StealEmoji Guildbank", code="S93bqTqKQ9rM"
|
||||
)
|
||||
except discord.HTTPException:
|
||||
await self.config.autobank.set(False)
|
||||
log.exception("Unable to create guilds, disabling autobank")
|
||||
return
|
||||
async with self.config.guildbanks() as guildbanks:
|
||||
guildbanks.append(guildbank.id)
|
||||
# Track generated guilds for easier deletion
|
||||
async with self.config.autobanked_guilds() as autobanked_guilds:
|
||||
autobanked_guilds.append(guildbank.id)
|
||||
|
||||
await asyncio.sleep(2)
|
||||
|
||||
if guildbank.text_channels:
|
||||
channel = guildbank.text_channels[0]
|
||||
else:
|
||||
# Always hits the else.
|
||||
# Maybe create_guild doesn't return guild object with
|
||||
# the template channel?
|
||||
channel = await guildbank.create_text_channel("invite-channel")
|
||||
invite = await channel.create_invite()
|
||||
|
||||
await self.bot.send_to_owners(invite)
|
||||
log.info(f"Guild created id {guildbank.id}. Invite: {invite}")
|
||||
else:
|
||||
if not await self.config.autobank():
|
||||
return
|
||||
|
||||
try:
|
||||
guildbank: discord.Guild = await self.bot.create_guild(
|
||||
"StealEmoji Guildbank", code="S93bqTqKQ9rM"
|
||||
)
|
||||
except discord.HTTPException:
|
||||
await self.config.autobank.set(False)
|
||||
log.exception("Unable to create guilds, disabling autobank")
|
||||
return
|
||||
async with self.config.guildbanks() as guildbanks:
|
||||
guildbanks.append(guildbank.id)
|
||||
# Track generated guilds for easier deletion
|
||||
async with self.config.autobanked_guilds() as autobanked_guilds:
|
||||
autobanked_guilds.append(guildbank.id)
|
||||
|
||||
await asyncio.sleep(2)
|
||||
|
||||
if guildbank.text_channels:
|
||||
channel = guildbank.text_channels[0]
|
||||
else:
|
||||
# Always hits the else.
|
||||
# Maybe create_guild doesn't return guild object with
|
||||
# the template channel?
|
||||
channel = await guildbank.create_text_channel("invite-channel")
|
||||
invite = await channel.create_invite()
|
||||
|
||||
await self.bot.send_to_owners(invite)
|
||||
log.info(f"Guild created id {guildbank.id}. Invite: {invite}")
|
||||
# Next, have I saved this emoji before (because uploaded emoji != orignal emoji)
|
||||
|
||||
if str(emoji.id) in await self.config.stolemoji():
|
||||
|
||||
@ -77,8 +77,7 @@ class Timerole(Cog):
|
||||
@commands.guild_only()
|
||||
async def timerole(self, ctx):
|
||||
"""Adjust timerole settings"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@timerole.command()
|
||||
async def addrole(
|
||||
@ -201,7 +200,7 @@ class Timerole(Cog):
|
||||
reapply = all_guilds[guild_id]["reapply"]
|
||||
role_dict = all_guilds[guild_id]["roles"]
|
||||
|
||||
if not any(role_data for role_data in role_dict.values()): # No roles
|
||||
if not any(role_dict.values()): # No roles
|
||||
log.debug(f"No roles are configured for guild: {guild}")
|
||||
continue
|
||||
|
||||
@ -232,7 +231,7 @@ class Timerole(Cog):
|
||||
log.debug(f"{member.display_name} - Not time to check again yet")
|
||||
continue
|
||||
member: discord.Member
|
||||
has_roles = set(r.id for r in member.roles)
|
||||
has_roles = {r.id for r in member.roles}
|
||||
|
||||
# Stop if they currently have or don't have the role, and mark had_role
|
||||
if (int(role_id) in has_roles and not role_data["remove"]) or (
|
||||
|
||||
@ -19,8 +19,7 @@ class Unicode(Cog):
|
||||
@commands.group(name="unicode", pass_context=True)
|
||||
async def unicode(self, ctx):
|
||||
"""Encode/Decode a Unicode character."""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@unicode.command()
|
||||
async def decode(self, ctx: commands.Context, character):
|
||||
|
||||
@ -90,7 +90,7 @@ async def parse_code(code, game):
|
||||
if len(built) < digits:
|
||||
built += c
|
||||
|
||||
if built == "T" or built == "W" or built == "N":
|
||||
if built in ["T", "W", "N"]:
|
||||
# Random Towns
|
||||
category = built
|
||||
built = ""
|
||||
@ -116,8 +116,6 @@ async def parse_code(code, game):
|
||||
options = [role for role in ROLE_LIST if 10 + idx in role.category]
|
||||
elif category == "N":
|
||||
options = [role for role in ROLE_LIST if 20 + idx in role.category]
|
||||
pass
|
||||
|
||||
if not options:
|
||||
raise IndexError("No Match Found")
|
||||
|
||||
@ -130,11 +128,8 @@ async def parse_code(code, game):
|
||||
|
||||
async def encode(role_list, rand_roles):
|
||||
"""Convert role list to code"""
|
||||
out_code = ""
|
||||
|
||||
digit_sort = sorted(role for role in role_list if role < 10)
|
||||
for role in digit_sort:
|
||||
out_code += str(role)
|
||||
out_code = "".join(str(role) for role in digit_sort)
|
||||
|
||||
digit_sort = sorted(role for role in role_list if 10 <= role < 100)
|
||||
if digit_sort:
|
||||
|
||||
@ -526,9 +526,10 @@ class Game:
|
||||
|
||||
async def _notify(self, event_name, **kwargs):
|
||||
for i in range(1, 7): # action guide 1-6 (0 is no action)
|
||||
tasks = []
|
||||
for event in self.listeners.get(event_name, {}).get(i, []):
|
||||
tasks.append(asyncio.create_task(event(**kwargs)))
|
||||
tasks = [
|
||||
asyncio.create_task(event(**kwargs))
|
||||
for event in self.listeners.get(event_name, {}).get(i, [])
|
||||
]
|
||||
|
||||
# Run same-priority task simultaneously
|
||||
await asyncio.gather(*tasks)
|
||||
@ -555,10 +556,7 @@ class Game:
|
||||
async def generate_targets(self, channel, with_roles=False):
|
||||
embed = discord.Embed(title="Remaining Players", description="[ID] - [Name]")
|
||||
for i, player in enumerate(self.players):
|
||||
if player.alive:
|
||||
status = ""
|
||||
else:
|
||||
status = "*[Dead]*-"
|
||||
status = "" if player.alive else "*[Dead]*-"
|
||||
if with_roles or not player.alive:
|
||||
embed.add_field(
|
||||
name=f"{i} - {status}{player.member.display_name}",
|
||||
@ -579,7 +577,7 @@ class Game:
|
||||
if channel_id not in self.p_channels:
|
||||
self.p_channels[channel_id] = self.default_secret_channel.copy()
|
||||
|
||||
for x in range(10): # Retry 10 times
|
||||
for _ in range(10): # Retry 10 times
|
||||
try:
|
||||
await asyncio.sleep(1) # This will have multiple calls
|
||||
self.p_channels[channel_id]["players"].append(role.player)
|
||||
@ -706,9 +704,7 @@ class Game:
|
||||
if not self.any_votes_remaining:
|
||||
await channel.send("Voting is not allowed right now")
|
||||
return
|
||||
elif channel.name in self.p_channels:
|
||||
pass
|
||||
else:
|
||||
elif channel.name not in self.p_channels:
|
||||
# Not part of the game
|
||||
await channel.send("Cannot vote in this channel")
|
||||
return
|
||||
@ -757,14 +753,14 @@ class Game:
|
||||
await self._at_voted(target)
|
||||
|
||||
async def eval_results(self, target, source=None, method=None):
|
||||
if method is not None:
|
||||
out = "**{ID}** - " + method
|
||||
return out.format(ID=target.id, target=target.member.display_name)
|
||||
else:
|
||||
if method is None:
|
||||
return "**{ID}** - {target} the {role} was found dead".format(
|
||||
ID=target.id, target=target.member.display_name, role=await target.role.get_role()
|
||||
)
|
||||
|
||||
out = "**{ID}** - " + method
|
||||
return out.format(ID=target.id, target=target.member.display_name)
|
||||
|
||||
async def _quit(self, player):
|
||||
"""
|
||||
Have player quit the game
|
||||
|
||||
@ -75,8 +75,7 @@ class Werewolf(Cog):
|
||||
"""
|
||||
Base command to adjust settings. Check help for command list.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@commands.guild_only()
|
||||
@wwset.command(name="list")
|
||||
@ -166,8 +165,7 @@ class Werewolf(Cog):
|
||||
"""
|
||||
Base command for this cog. Check help for the commands list.
|
||||
"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
pass
|
||||
pass
|
||||
|
||||
@commands.guild_only()
|
||||
@ww.command(name="new")
|
||||
@ -348,8 +346,7 @@ class Werewolf(Cog):
|
||||
"""
|
||||
Find custom roles by name, alignment, category, or ID
|
||||
"""
|
||||
if ctx.invoked_subcommand is None or ctx.invoked_subcommand == self.ww_search:
|
||||
pass
|
||||
pass
|
||||
|
||||
@ww_search.command(name="name")
|
||||
async def ww_search_name(self, ctx: commands.Context, *, name):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user