Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@commands.command()
async def xpbetween(self, ctx, start_level : int, end_level : int):
start_xp = sum(int(level + 300 * 2 ** (level / 7)) for level in range(1, start_level))
end_xp = (start_xp + sum(int(level + 300 * 2 ** (level / 7)) for level in range(start_level, end_level))) // 4
start_xp //= 4
await ctx.send(f"{end_xp - start_xp:,} xp between level {start_level} and level {end_level}")
def _populate_entries(*channels: Union[str, int]):
names = set()
ids = set()
for channel in channels:
if isinstance(channel, str):
if channel.isdigit():
# Handle ids in the string form
ids.add(int(channel))
else:
names.add(channel)
elif isinstance(channel, int):
ids.add(str(channel))
if len(names | ids) > 100:
raise HTTPException('Bad Request - Total entries must not exceed 100.')
return names, ids
except asyncio.InvalidStateError:
pass
if fut.done():
futures.remove(fut)
if futures:
await asyncio.wait(futures)
await self._dispatch('ready')
self.is_ready.set()
elif data == ':tmi.twitch.tv NOTICE * :Login authentication failed' or\
data == ':tmi.twitch.tv NOTICE * :Improperly formatted auth':
log.warning('Authentication failed | %s', self._token)
raise AuthenticationError('Websocket Authentication Failure... Check your token and nick.')
_groupsdict = {}
if data.startswith("PING"):
match = self.regex["ping"]
else:
match = self.regex["data"]
result = match.match(data)
badges = self.regex['badges'].match(data)
if badges:
badges = {'name': badges.group('name'), 'mod': badges.group('mod'), 'action': badges.group('action'),
'channel': badges.group('channel')}
from twitchio.ext import commands
@commands.cog()
class Interactions:
def __init__(self, bot):
self.bot = bot
@commands.command(aliases = ("goodbye",))
async def bye(self, ctx, *, user = None):
if not user or user.lower() == "harmonbot":
await ctx.send(f"Bye, {ctx.author.name.capitalize()}!")
else:
await ctx.send(f"{user.title().lstrip('/')}, {ctx.author.name.capitalize()} says goodbye!")
@commands.command(aliases = ("hi",))
async def hello(self, ctx, *, user = None):
if not user or user.lower() == "harmonbot":
await ctx.send(f"Hello, {ctx.author.name.capitalize()}!")
@commands.command()
async def forecast(self, ctx, *, location = ""):
# TODO: Detailed forecast option?
if not location or location.lower() == ctx.channel.name:
location = await self.bot.db.fetchval("SELECT location FROM twitch.locations WHERE channel = $1", ctx.channel.name)
if not location:
return await ctx.send(f"Error: Location not specified")
try:
forecaster = self.bot.owm_client.daily_forecast(location)
except (pyowm.exceptions.api_response_error.NotFoundError,
pyowm.exceptions.api_call_error.BadGatewayError) as e:
# TODO: Catch base exceptions?
return await ctx.send(f"Error: {e}")
forecast = forecaster.get_forecast()
location = forecast.get_location()
output = f"{location.get_name()}, {location.get_country()}"
for weather in forecast:
@commands.command()
async def averagefps(self, ctx):
users = await self.bot.get_users(ctx.channel.name)
url = "https://api.twitch.tv/kraken/streams/" + users[0].id
params = {"client_id": self.bot.http.client_id}
headers = {"Accept": "application/vnd.twitchtv.v5+json"}
async with self.bot.aiohttp_session.get(url, params = params, headers = headers) as resp:
data = await resp.json()
stream = data.get("stream")
if not stream:
return await ctx.send("Average FPS not found.")
await ctx.send(f"Average FPS: {stream['average_fps']}")
@commands.command()
async def time(self, ctx, *, location = ""):
if not location or location.lower() == ctx.channel.name:
location = await self.bot.db.fetchval("SELECT location FROM twitch.locations WHERE channel = $1", ctx.channel.name)
if not location:
return await ctx.send(f"Error: Location not specified")
try:
geocode_data = await get_geocode_data(location, aiohttp_session = self.bot.aiohttp_session)
latitude = geocode_data["geometry"]["location"]["lat"]
longitude = geocode_data["geometry"]["location"]["lng"]
timezone_data = await get_timezone_data(latitude = latitude, longitude = longitude,
aiohttp_session = self.bot.aiohttp_session)
except UnitOutputError as e:
return await ctx.send(f"Error: {e}")
location_time = datetime.datetime.now(datetime.timezone(datetime.timedelta(
seconds = timezone_data["dstOffset"] + timezone_data["rawOffset"])))
# TODO: Use method for Discord time command
@commands.command()
async def define(self, ctx, *, word):
url = f"http://api.wordnik.com:80/v4/word.json/{word}/definitions"
params = {"limit": 1, "includeRelated": "false", "useCanonical": "false", "includeTags": "false",
"api_key": self.bot.WORDNIK_API_KEY}
async with self.bot.aiohttp_session.get(url, params = params) as resp:
if resp.status == 404:
return await ctx.send("Error: Not found")
data = await resp.json()
if not data:
return await ctx.send("Definition not found.")
await ctx.send(f"{data[0]['word']}: {data[0]['text']}")
import os
import sys
import aiohttp
import asyncpg
import dotenv
import pyowm
from utilities import context
from utilities import logging
sys.path.insert(0, "..")
from units.games import eightball
sys.path.pop(0)
class Bot(commands.Bot):
def __init__(self, loop = None, initial_channels = None, **kwargs):
self.version = "3.0.0-b.133"
loop = loop or asyncio.get_event_loop()
if initial_channels is None:
initial_channels = []
initial_channels = list(initial_channels)
# Constants
self.char_limit = self.character_limit = 500
# aiohttp Client Session - initialized on ready
self.aiohttp_session = None
# Credentials
@commands.command(aliases = ("goodbye",))
async def bye(self, ctx, *, user = None):
if not user or user.lower() == "harmonbot":
await ctx.send(f"Bye, {ctx.author.name.capitalize()}!")
else:
await ctx.send(f"{user.title().lstrip('/')}, {ctx.author.name.capitalize()} says goodbye!")