diff --git a/.idea/.gitignore b/.idea/.gitignore
new file mode 100644
index 0000000..26d3352
--- /dev/null
+++ b/.idea/.gitignore
@@ -0,0 +1,3 @@
+# Default ignored files
+/shelf/
+/workspace.xml
diff --git a/.idea/Ava-Cogs.iml b/.idea/Ava-Cogs.iml
new file mode 100644
index 0000000..d6ebd48
--- /dev/null
+++ b/.idea/Ava-Cogs.iml
@@ -0,0 +1,9 @@
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/discord.xml b/.idea/discord.xml
new file mode 100644
index 0000000..912db82
--- /dev/null
+++ b/.idea/discord.xml
@@ -0,0 +1,14 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/misc.xml b/.idea/misc.xml
new file mode 100644
index 0000000..6f29fee
--- /dev/null
+++ b/.idea/misc.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/modules.xml b/.idea/modules.xml
new file mode 100644
index 0000000..37313d7
--- /dev/null
+++ b/.idea/modules.xml
@@ -0,0 +1,8 @@
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/vcs.xml b/.idea/vcs.xml
new file mode 100644
index 0000000..35eb1dd
--- /dev/null
+++ b/.idea/vcs.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/lastfm/__init__.py b/lastfm/__init__.py
new file mode 100644
index 0000000..9873a82
--- /dev/null
+++ b/lastfm/__init__.py
@@ -0,0 +1,9 @@
+from .lastfm import LastFM
+
+__red_end_user_data_statement__ = "This cog stores a user's last.fm username, their amount of VC scrobbles, and a last.fm session key to scrobble for them. It also stores the user's crowns. This is all data that can be cleared."
+
+
+async def setup(bot):
+ cog = LastFM(bot)
+ await bot.add_cog(cog)
+ await cog.initialize()
diff --git a/lastfm/abc.py b/lastfm/abc.py
new file mode 100644
index 0000000..69a08c9
--- /dev/null
+++ b/lastfm/abc.py
@@ -0,0 +1,22 @@
+from abc import ABC
+
+from redbot.core import Config, commands
+from redbot.core.bot import Red
+
+
+class CompositeMetaClass(type(commands.Cog), type(ABC)):
+ """
+ This allows the metaclass used for proper type detection to coexist with discord.py's
+ metaclass.
+ """
+
+
+class MixinMeta(ABC):
+ """
+ Base class for well behaved type hint detection with composite class.
+ Basically, to keep developers sane when not all attributes are defined in each mixin.
+ """
+
+ def __init__(self, *_args):
+ self.config: Config
+ self.bot: Red
diff --git a/lastfm/charts.py b/lastfm/charts.py
new file mode 100644
index 0000000..6875f99
--- /dev/null
+++ b/lastfm/charts.py
@@ -0,0 +1,421 @@
+import asyncio
+from io import BytesIO
+
+import discord
+from PIL import Image, ImageDraw, ImageFile, ImageFont
+from redbot.core import commands
+from redbot.core.utils import AsyncIter
+from redbot.core.utils.chat_formatting import escape
+
+from .abc import MixinMeta
+from .exceptions import *
+from .fmmixin import FMMixin
+
+NO_IMAGE_PLACEHOLDER = (
+ "https://lastfm.freetls.fastly.net/i/u/300x300/2a96cbd8b46e442fc41c2b86b821562f.png"
+)
+ImageFile.LOAD_TRUNCATED_IMAGES = True
+
+command_fm = FMMixin.command_fm
+command_fm_server = FMMixin.command_fm_server
+
+
+class ChartMixin(MixinMeta):
+ """Chart Commands"""
+
+ async def get_img(self, url):
+ async with self.session.get(url or NO_IMAGE_PLACEHOLDER) as resp:
+ if resp.status == 200:
+ img = await resp.read()
+ return img
+ async with self.session.get(NO_IMAGE_PLACEHOLDER) as resp:
+ img = await resp.read()
+ return img
+
+ @command_fm.command(
+ name="chart", usage="[album | artist | recent | track] [timeframe] [width]x[height]"
+ )
+ @commands.max_concurrency(1, commands.BucketType.user)
+ async def command_chart(self, ctx, *args):
+ """
+ Visual chart of your top albums, tracks or artists.
+
+ Defaults to top albums, weekly, 3x3.
+ """
+ conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(conf)
+ arguments = self.parse_chart_arguments(args)
+ if arguments["width"] + arguments["height"] > 31: # TODO: Figure out a reasonable value.
+ return await ctx.send(
+ "Size is too big! Chart `width` + `height` total must not exceed `31`"
+ )
+ msg = await ctx.send("Gathering images and data, this may take some time.")
+ data = await self.api_request(
+ ctx,
+ {
+ "user": conf["lastfm_username"],
+ "method": arguments["method"],
+ "period": arguments["period"],
+ "limit": arguments["amount"],
+ },
+ )
+ chart = []
+ chart_type = "ERROR"
+ async with ctx.typing():
+ if arguments["method"] == "user.gettopalbums":
+ chart_type = "top album"
+ albums = data["topalbums"]["album"]
+ async for album in AsyncIter(albums[: arguments["width"] * arguments["height"]]):
+ name = album["name"]
+ artist = album["artist"]["name"]
+ plays = album["playcount"]
+ if album["image"][3]["#text"] in self.chart_data:
+ chart_img = self.chart_data[album["image"][3]["#text"]]
+ else:
+ chart_img = await self.get_img(album["image"][3]["#text"])
+ self.chart_data[album["image"][3]["#text"]] = chart_img
+ chart.append(
+ (
+ f"{plays} {self.format_plays(plays)}\n{name} - {artist}",
+ chart_img,
+ )
+ )
+ img = await self.bot.loop.run_in_executor(
+ None,
+ charts,
+ chart,
+ arguments["width"],
+ arguments["height"],
+ self.data_loc,
+ )
+
+ elif arguments["method"] == "user.gettopartists":
+ chart_type = "top artist"
+ artists = data["topartists"]["artist"]
+ if self.login_token:
+ scraped_images = await self.scrape_artists_for_chart(
+ ctx, conf["lastfm_username"], arguments["period"], arguments["amount"]
+ )
+ else:
+ scraped_images = [NO_IMAGE_PLACEHOLDER] * arguments["amount"]
+ iterator = AsyncIter(artists[: arguments["width"] * arguments["height"]])
+ async for i, artist in iterator.enumerate():
+ name = artist["name"]
+ plays = artist["playcount"]
+ if i < len(scraped_images) and scraped_images[i] in self.chart_data:
+ chart_img = self.chart_data[scraped_images[i]]
+ else:
+ chart_img = await self.get_img(scraped_images[i])
+ self.chart_data[scraped_images[i]] = chart_img
+ chart.append(
+ (
+ f"{plays} {self.format_plays(plays)}\n{name}",
+ chart_img,
+ )
+ )
+ img = await self.bot.loop.run_in_executor(
+ None,
+ charts,
+ chart,
+ arguments["width"],
+ arguments["height"],
+ self.data_loc,
+ )
+
+ elif arguments["method"] == "user.getrecenttracks":
+ chart_type = "recent tracks"
+ tracks = data["recenttracks"]["track"]
+ if isinstance(tracks, dict):
+ tracks = [tracks]
+ async for track in AsyncIter(tracks[: arguments["width"] * arguments["height"]]):
+ name = track["name"]
+ artist = track["artist"]["#text"]
+ if track["image"][3]["#text"] in self.chart_data:
+ chart_img = self.chart_data[track["image"][3]["#text"]]
+ else:
+ chart_img = await self.get_img(track["image"][3]["#text"])
+ self.chart_data[track["image"][3]["#text"]] = chart_img
+ chart.append(
+ (
+ f"{name} - {artist}",
+ chart_img,
+ )
+ )
+ img = await self.bot.loop.run_in_executor(
+ None,
+ track_chart,
+ chart,
+ arguments["width"],
+ arguments["height"],
+ self.data_loc,
+ )
+ elif arguments["method"] == "user.gettoptracks":
+ chart_type = "top tracks"
+ tracks = data["toptracks"]["track"]
+ scraped_images = await self.scrape_artists_for_chart(
+ ctx, conf["lastfm_username"], arguments["period"], arguments["amount"]
+ )
+ if isinstance(tracks, dict):
+ tracks = [tracks]
+ async for track in AsyncIter(tracks[: arguments["width"] * arguments["height"]]):
+ name = track["name"]
+ artist = track["artist"]["name"]
+ plays = track["playcount"]
+ if name in self.chart_data:
+ chart_img = self.chart_data[name]
+ else:
+ chart_img = await self.get_img(await self.scrape_artist_image(artist, ctx))
+ self.chart_data[name] = chart_img
+ chart.append(
+ (
+ f"{plays} {self.format_plays(plays)}\n{name} - {artist}",
+ chart_img,
+ )
+ )
+ img = await self.bot.loop.run_in_executor(
+ None,
+ charts,
+ chart,
+ arguments["width"],
+ arguments["height"],
+ self.data_loc,
+ )
+ await msg.delete()
+ u = conf["lastfm_username"]
+ try:
+ await ctx.send(
+ f"`{u} - {self.humanized_period(arguments['period'])} - {arguments['width']}x{arguments['height']} {chart_type} chart`",
+ file=img,
+ )
+ except discord.HTTPException:
+ await ctx.send("File is to big to send, try lowering the size.")
+
+ @command_fm_server.command(
+ name="chart", usage="[album | artist | tracks] [timeframe] [width]x[height]"
+ )
+ @commands.max_concurrency(1, commands.BucketType.user)
+ async def server_chart(self, ctx, *args):
+ """Visual chart of the servers albums, artists or tracks."""
+ arguments = self.parse_chart_arguments(args)
+ if arguments["width"] + arguments["height"] > 31: # TODO: Figure out a reasonable value.
+ return await ctx.send(
+ "Size is too big! Chart `width` + `height` total must not exceed `31`"
+ )
+ if arguments["method"] not in [
+ "user.gettopalbums",
+ "user.gettopartists",
+ "user.gettoptracks",
+ ]:
+ return await ctx.send("Only albums, artists and tracks are supported.")
+ chart_total = arguments["width"] * arguments["height"]
+ msg = await ctx.send("Gathering images and data, this may take some time.")
+ tasks = []
+ userlist = await self.config.all_users()
+ guildusers = [x.id for x in ctx.guild.members]
+ userslist = [user for user in userlist if user in guildusers]
+ datatype = {
+ "user.gettopalbums": "album",
+ "user.gettopartists": "artist",
+ "user.gettoptracks": "track",
+ }
+ for user in userslist:
+ lastfm_username = userlist[user]["lastfm_username"]
+ if lastfm_username is None:
+ continue
+ member = ctx.guild.get_member(user)
+ if member is None:
+ continue
+
+ tasks.append(
+ self.get_server_top(
+ ctx,
+ lastfm_username,
+ datatype.get(arguments["method"]),
+ arguments["period"],
+ arguments["amount"],
+ )
+ )
+ chart = []
+ chart_type = "ERROR"
+ if not tasks:
+ return await ctx.send("No users have set their last.fm username yet.")
+ content_map = {}
+ async with ctx.typing():
+ data = await asyncio.gather(*tasks)
+ if arguments["method"] == "user.gettopalbums":
+ chart_type = "top album"
+ for user_data in data:
+ if user_data is None:
+ continue
+ for album in user_data:
+ album_name = album["name"]
+ artist = album["artist"]["name"]
+ name = f"{album_name} — {artist}"
+ plays = int(album["playcount"])
+ if name in content_map:
+ content_map[name]["plays"] += plays
+ else:
+ content_map[name] = {
+ "plays": plays,
+ "link": album["image"][3]["#text"],
+ }
+ elif arguments["method"] == "user.gettopartists":
+ chart_type = "top artist"
+ for user_data in data:
+ if user_data is None:
+ continue
+ for artist in user_data:
+ name = artist["name"]
+ plays = int(artist["playcount"])
+ if name in content_map:
+ content_map[name]["plays"] += plays
+ else:
+ content_map[name] = {"plays": plays}
+ elif arguments["method"] == "user.gettoptracks":
+ chart_type = "top tracks"
+ for user in data:
+ if user is None:
+ continue
+ for user_data in user:
+ name = f'{escape(user_data["artist"]["name"])} — *{escape(user_data["name"])}*'
+ plays = int(user_data["playcount"])
+ if name in content_map:
+ content_map[name]["plays"] += plays
+ else:
+ content_map[name] = {
+ "plays": plays,
+ "link": user_data["artist"]["name"],
+ }
+ cached_images = {}
+ for i, (name, content_data) in enumerate(
+ sorted(content_map.items(), key=lambda x: x[1]["plays"], reverse=True), start=1
+ ):
+ if arguments["method"] == "user.gettopartists":
+ image = await self.get_img(await self.scrape_artist_image(name, ctx))
+ elif arguments["method"] == "user.gettoptracks":
+ if content_data["link"] in cached_images:
+ image = cached_images[content_data["link"]]
+ else:
+ image = await self.get_img(
+ await self.scrape_artist_image(content_data["link"], ctx)
+ )
+ cached_images[content_data["link"]] = image
+ else:
+ image = await self.get_img(content_data["link"])
+ chart.append(
+ (
+ f"{content_data['plays']} {self.format_plays(content_data['plays'])}\n{name}",
+ image,
+ )
+ )
+ if i >= chart_total:
+ break
+ img = await self.bot.loop.run_in_executor(
+ None,
+ charts,
+ chart,
+ arguments["width"],
+ arguments["height"],
+ self.data_loc,
+ )
+ await msg.delete()
+ try:
+ await ctx.send(
+ f"`{ctx.guild} - {self.humanized_period(arguments['period'])} - {arguments['width']}x{arguments['height']} {chart_type} chart`",
+ file=img,
+ )
+ except discord.HTTPException:
+ await ctx.send("File is to big to send, try lowering the size.")
+
+
+def charts(data, w, h, loc):
+ fnt_file = f"{loc}/fonts/Arial Unicode.ttf"
+ fnt = ImageFont.truetype(fnt_file, 18, encoding="utf-8")
+ imgs = []
+ for item in data:
+ img = BytesIO(item[1])
+ image = Image.open(img).convert("RGBA")
+ draw = ImageDraw.Draw(image)
+ texts = item[0].split("\n")
+ if len(texts[1]) > 30:
+ height = 223
+ text = f"{texts[0]}\n{texts[1][:30]}\n{texts[1][30:]}"
+ else:
+ height = 247
+ text = item[0]
+ draw.text(
+ (5, height),
+ text,
+ fill=(255, 255, 255, 255),
+ font=fnt,
+ stroke_width=1,
+ stroke_fill=(0, 0, 0),
+ )
+ _file = BytesIO()
+ image.save(_file, "png")
+ _file.name = f"{item[0]}.png"
+ _file.seek(0)
+ imgs.append(_file)
+ return create_graph(imgs, w, h)
+
+
+def track_chart(data, w, h, loc):
+ fnt_file = f"{loc}/fonts/Arial Unicode.ttf"
+ fnt = ImageFont.truetype(fnt_file, 18, encoding="utf-8")
+ imgs = []
+ for item in data:
+ img = BytesIO(item[1])
+ image = Image.open(img).convert("RGBA")
+ draw = ImageDraw.Draw(image)
+ if len(item[0]) > 30:
+ height = 243
+ text = f"{item[0][:30]}\n{item[0][30:]}"
+ else:
+ height = 267
+ text = item[0]
+ draw.text(
+ (5, height),
+ text,
+ fill=(255, 255, 255, 255),
+ font=fnt,
+ stroke_width=1,
+ stroke_fill=(0, 0, 0),
+ )
+ _file = BytesIO()
+ image.save(_file, "png")
+ _file.name = f"{item[0]}.png"
+ _file.seek(0)
+ imgs.append(_file)
+ return create_graph(imgs, w, h)
+
+
+def chunks(l, n):
+ """Yield successive n-sized chunks from l."""
+ for i in range(0, len(l), n):
+ yield l[i : i + n]
+
+
+def create_graph(data, w, h):
+ dimensions = (300 * w, 300 * h)
+ final = Image.new("RGBA", dimensions)
+ images = chunks(data, w)
+ y = 0
+ for chunked in images:
+ x = 0
+ for img in chunked:
+ new = Image.open(img)
+ w, h = new.size
+ final.paste(new, (x, y, x + w, y + h))
+ x += 300
+ y += 300
+ w, h = final.size
+ if w > 2100 and h > 2100:
+ final = final.resize(
+ (2100, 2100), resample=Image.ANTIALIAS
+ ) # Resize cause a 6x6k image is blocking when being sent
+ file = BytesIO()
+ final.save(file, "webp")
+ file.name = f"chart.webp"
+ file.seek(0)
+ image = discord.File(file)
+ return image
diff --git a/lastfm/compare.py b/lastfm/compare.py
new file mode 100644
index 0000000..4e211fd
--- /dev/null
+++ b/lastfm/compare.py
@@ -0,0 +1,295 @@
+import contextlib
+from io import BytesIO
+
+import discord
+import tabulate
+from PIL import Image, ImageDraw, ImageFont
+from redbot.core.utils import AsyncIter
+from redbot.core.utils.chat_formatting import humanize_number
+
+from .abc import MixinMeta
+from .exceptions import *
+from .fmmixin import FMMixin
+
+command_fm = FMMixin.command_fm
+command_fm_server = FMMixin.command_fm_server
+
+
+class CompareMixin(MixinMeta):
+ """Commands for comparing two users"""
+
+ def make_table_into_image(self, text):
+
+ color = (150, 123, 182)
+
+ lines = 0
+ keep_going = True
+ width = 0
+ for line in text:
+ if keep_going:
+ width += 6.6
+ if line == "\n":
+ lines += 16.5
+ keep_going = False
+
+ img = Image.new("RGBA", (int(width), int(lines)), color=(255, 0, 0, 0))
+
+ d = ImageDraw.Draw(img)
+ fnt_file = f"{self.data_loc}/fonts/NotoSansMono-Regular.ttf"
+ font = ImageFont.truetype(fnt_file, 11, encoding="utf-8")
+ d.text((0, 0), text, fill=color, font=font)
+
+ final = BytesIO()
+ img.save(final, "webp")
+ final.seek(0)
+ return discord.File(final, "result.webp")
+
+ @command_fm.group(name="compare")
+ async def command_compare(self, ctx):
+ """Compare two users music tastes"""
+
+ @command_compare.command(name="artists", aliases=["artist"])
+ async def compare_artists(self, ctx, user: discord.Member, period: str = "1month"):
+ """
+ Compare your top artists with someone else.
+
+ `[period]` can be one of: overall, 7day, 1month, 3month, 6month, 12month
+ The default is 1 month.
+ """
+ if user == ctx.author:
+ await ctx.send("You need to compare with someone else.")
+ return
+
+ period, displayperiod = self.get_period(period)
+ if not period:
+ await ctx.send(
+ "Invalid period. Valid periods are: overall, 7day, 1month, 3month, 6month, 12month"
+ )
+ return
+
+ author_conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(author_conf)
+ user_conf = await self.config.user(user).all()
+ self.check_if_logged_in(user_conf, False)
+ async with ctx.typing():
+ author_data = await self.api_request(
+ ctx,
+ {
+ "user": author_conf["lastfm_username"],
+ "method": "user.gettopartists",
+ "period": period,
+ "limit": "10",
+ },
+ )
+ author_artists = author_data["topartists"]["artist"]
+ if not author_artists:
+ if period == "overall":
+ await ctx.send("You haven't listened to any artists yet.")
+ else:
+ await ctx.send(
+ "You haven't listened to any artists in the last {}s.".format(period)
+ )
+ return
+
+ g = await ctx.send("Gathering data... This might take a while.")
+
+ author_plays = []
+ artist_names = []
+ for artist in author_artists:
+ if artist["playcount"] == 1:
+ author_plays.append(f"{artist['playcount']} Play")
+ else:
+ author_plays.append(f"{humanize_number(artist['playcount'])} Plays")
+ artist_names.append(artist["name"])
+
+ user_plays = []
+ async for artist in AsyncIter(author_artists):
+ plays = await self.get_playcount(
+ ctx, user_conf["lastfm_username"], artist["name"], period
+ )
+ if plays == 1:
+ user_plays.append(f"{plays} Play")
+ else:
+ user_plays.append(f"{humanize_number(plays)} Plays")
+
+ data = {"Artist": artist_names, ctx.author: author_plays, user: user_plays}
+ table = tabulate.tabulate(data, headers="keys", tablefmt="fancy_grid")
+ color = await ctx.embed_colour()
+ img = await self.bot.loop.run_in_executor(None, self.make_table_into_image, table)
+ embed = discord.Embed(color=color, title=f"{ctx.author} vs {user} ({displayperiod})")
+ embed.set_image(url="attachment://result.webp")
+
+ with contextlib.suppress(discord.NotFound):
+ await g.delete()
+
+ await ctx.send(file=img, embed=embed)
+
+ @command_compare.command(name="tracks", aliases=["track"])
+ async def compare_tracks(self, ctx, user: discord.Member, period: str = "1month"):
+ """
+ Compare your top tracks with someone else.
+
+ `[period]` can be one of: overall, 7day, 1month, 3month, 6month, 12month
+ The default is 1 month.
+ """
+ if user == ctx.author:
+ await ctx.send("You need to compare with someone else.")
+ return
+
+ period, displayperiod = self.get_period(period)
+ if not period:
+ await ctx.send(
+ "Invalid period. Valid periods are: overall, 7day, 1month, 3month, 6month, 12month"
+ )
+ return
+
+ author_conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(author_conf)
+ user_conf = await self.config.user(user).all()
+ self.check_if_logged_in(user_conf, False)
+ async with ctx.typing():
+ author_data = await self.api_request(
+ ctx,
+ {
+ "user": author_conf["lastfm_username"],
+ "method": "user.gettoptracks",
+ "period": period,
+ "limit": "10",
+ },
+ )
+ author_tracks = author_data["toptracks"]["track"]
+ if not author_tracks:
+ if period == "overall":
+ await ctx.send("You haven't listened to any tracks yet.")
+ else:
+ await ctx.send("You haven't listened to any tracks in that time period.")
+ return
+
+ g = await ctx.send("Gathering data... This might take a while.")
+
+ author_plays = []
+ artist_names = []
+ track_names = []
+ for track in author_tracks:
+ if track["playcount"] == 1:
+ author_plays.append(f"{track['playcount']} Play")
+ else:
+ author_plays.append(f"{humanize_number(track['playcount'])} Plays")
+ artist_names.append(track["artist"]["name"])
+ track_names.append(track["name"])
+
+ user_plays = []
+ async for track in AsyncIter(author_tracks):
+ plays = await self.get_playcount_track(
+ ctx,
+ user_conf["lastfm_username"],
+ track["artist"]["name"],
+ track["name"],
+ period,
+ )
+ if plays == 1:
+ user_plays.append(f"{plays} Play")
+ else:
+ user_plays.append(f"{humanize_number(plays)} Plays")
+
+ data = {
+ "Artist": artist_names,
+ "Track": track_names,
+ ctx.author: author_plays,
+ user: user_plays,
+ }
+ table = tabulate.tabulate(data, headers="keys", tablefmt="fancy_grid")
+ color = await ctx.embed_colour()
+ img = await self.bot.loop.run_in_executor(None, self.make_table_into_image, table)
+ embed = discord.Embed(color=color, title=f"{ctx.author} vs {user} ({displayperiod})")
+ embed.set_image(url="attachment://result.webp")
+
+ with contextlib.suppress(discord.NotFound):
+ await g.delete()
+
+ await ctx.send(file=img, embed=embed)
+
+ @command_compare.command(name="albums", aliases=["album"])
+ async def compare_albums(self, ctx, user: discord.Member, period: str = "1month"):
+ """
+ Compare your top albums with someone else.
+
+ `[period]` can be one of: overall, 7day, 1month, 3month, 6month, 12month
+ The default is 1 month.
+ """
+ if user == ctx.author:
+ await ctx.send("You need to compare with someone else.")
+ return
+
+ period, displayperiod = self.get_period(period)
+ if not period:
+ await ctx.send(
+ "Invalid period. Valid periods are: overall, 7day, 1month, 3month, 6month, 12month"
+ )
+ return
+
+ author_conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(author_conf)
+ user_conf = await self.config.user(user).all()
+ self.check_if_logged_in(user_conf, False)
+ async with ctx.typing():
+ author_data = await self.api_request(
+ ctx,
+ {
+ "user": author_conf["lastfm_username"],
+ "method": "user.gettopalbums",
+ "period": period,
+ "limit": "10",
+ },
+ )
+ author_albums = author_data["topalbums"]["album"]
+ if not author_albums:
+ if period == "overall":
+ await ctx.send("You haven't listened to any albums yet.")
+ else:
+ await ctx.send("You haven't listened to any albums in that time period.")
+ return
+
+ g = await ctx.send("Gathering data... This might take a while.")
+
+ author_plays = []
+ artist_names = []
+ album_names = []
+ for album in author_albums:
+ if album["playcount"] == 1:
+ author_plays.append(f"{album['playcount']} Play")
+ else:
+ author_plays.append(f"{humanize_number(album['playcount'])} Plays")
+ artist_names.append(album["artist"]["name"])
+ album_names.append(album["name"])
+
+ user_plays = []
+ async for album in AsyncIter(author_albums):
+ plays = await self.get_playcount_album(
+ ctx,
+ user_conf["lastfm_username"],
+ album["artist"]["name"],
+ album["name"],
+ period,
+ )
+ if plays == 1:
+ user_plays.append(f"{plays} Play")
+ else:
+ user_plays.append(f"{humanize_number(plays)} Plays")
+
+ data = {
+ "Artist": artist_names,
+ "Album": album_names,
+ ctx.author: author_plays,
+ user: user_plays,
+ }
+ table = tabulate.tabulate(data, headers="keys", tablefmt="fancy_grid")
+ color = await ctx.embed_colour()
+ img = await self.bot.loop.run_in_executor(None, self.make_table_into_image, table)
+ embed = discord.Embed(color=color, title=f"{ctx.author} vs {user} ({displayperiod})")
+ embed.set_image(url="attachment://result.webp")
+
+ with contextlib.suppress(discord.NotFound):
+ await g.delete()
+
+ await ctx.send(file=img, embed=embed)
diff --git a/lastfm/data/fonts/Arial Unicode.ttf b/lastfm/data/fonts/Arial Unicode.ttf
new file mode 100644
index 0000000..1537c5b
Binary files /dev/null and b/lastfm/data/fonts/Arial Unicode.ttf differ
diff --git a/lastfm/data/fonts/NotoSansMono-Regular.ttf b/lastfm/data/fonts/NotoSansMono-Regular.ttf
new file mode 100644
index 0000000..a850b21
Binary files /dev/null and b/lastfm/data/fonts/NotoSansMono-Regular.ttf differ
diff --git a/lastfm/exceptions.py b/lastfm/exceptions.py
new file mode 100644
index 0000000..7298a95
--- /dev/null
+++ b/lastfm/exceptions.py
@@ -0,0 +1,22 @@
+class LastFMError(Exception):
+ pass
+
+
+class NotLoggedInError(LastFMError):
+ pass
+
+
+class NeedToReauthorizeError(LastFMError):
+ pass
+
+
+class SilentDeAuthorizedError(LastFMError):
+ pass
+
+
+class NoScrobblesError(LastFMError):
+ pass
+
+
+class NotScrobblingError(LastFMError):
+ pass
diff --git a/lastfm/fmmixin.py b/lastfm/fmmixin.py
new file mode 100644
index 0000000..6896dc3
--- /dev/null
+++ b/lastfm/fmmixin.py
@@ -0,0 +1,24 @@
+from redbot.core import commands
+
+from .utils.tokencheck import tokencheck
+
+
+
+
+class FMMixin:
+ """This is mostly here to easily mess with things..."""
+
+
+ @commands.check(tokencheck)
+ @commands.group(name="fm")
+ async def command_fm(self, ctx: commands.Context):
+ """
+ LastFM commands
+ """
+
+
+ @command_fm.group(name="server")
+ async def command_fm_server(self, ctx: commands.Context):
+ """
+ LastFM Server Commands
+ """
diff --git a/lastfm/info.json b/lastfm/info.json
new file mode 100644
index 0000000..f1c1711
--- /dev/null
+++ b/lastfm/info.json
@@ -0,0 +1,27 @@
+{
+ "author": [
+ "flare(flare#0001)",
+ "joinem",
+ "fixator10",
+ "ryan"
+ ],
+ "install_msg": "Thanks for installing!\n\nOnce you've loaded the cog run the `[p]lastfmset` command for instructions on how to set your API key!",
+ "name": "LastFm",
+ "disabled": false,
+ "short": "Port of Miso Bot's LastFM Cog",
+ "description": "Port of Miso Bot's LastFM Cog,",
+ "tags": [
+ "LastFM",
+ "scrobbler",
+ "music"
+ ],
+ "hidden": false,
+ "requirements": [
+ "humanize",
+ "bs4",
+ "tabulate",
+ "wordcloud",
+ "arrow"
+ ],
+ "min_bot_version": "3.5.0"
+}
diff --git a/lastfm/lastfm.py b/lastfm/lastfm.py
new file mode 100644
index 0000000..d4171df
--- /dev/null
+++ b/lastfm/lastfm.py
@@ -0,0 +1,387 @@
+import asyncio
+import urllib.parse
+from operator import itemgetter
+
+import aiohttp
+import discord
+from redbot.core import Config, commands
+from redbot.core.data_manager import bundled_data_path
+from redbot.core.utils.chat_formatting import escape, pagify
+from redbot.core.utils.menus import DEFAULT_CONTROLS, menu
+
+from .abc import *
+from .charts import ChartMixin
+from .compare import CompareMixin
+from .exceptions import *
+from .fmmixin import FMMixin
+from .love import LoveMixin
+from .nowplaying import NowPlayingMixin
+from .profile import ProfileMixin
+from .recent import RecentMixin
+from .scrobbler import ScrobblerMixin
+from .tags import TagsMixin
+from .top import TopMixin
+from .utils.base import UtilsMixin
+from .utils.tokencheck import *
+from .whoknows import WhoKnowsMixin
+from .wordcloud import WordCloudMixin
+
+command_fm = FMMixin.command_fm
+
+
+class LastFM(
+ ChartMixin,
+ CompareMixin,
+ FMMixin,
+ LoveMixin,
+ NowPlayingMixin,
+ ProfileMixin,
+ RecentMixin,
+ ScrobblerMixin,
+ TagsMixin,
+ TopMixin,
+ UtilsMixin,
+ WhoKnowsMixin,
+ WordCloudMixin,
+ commands.Cog,
+ metaclass=CompositeMetaClass,
+):
+ """
+ Interacts with the last.fm API.
+ """
+
+ __version__ = "1.7.2"
+
+ # noinspection PyMissingConstructor
+ def __init__(self, bot, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.bot = bot
+ self.config = Config.get_conf(self, identifier=95932766180343808, force_registration=True)
+ defaults = {"lastfm_username": None, "session_key": None, "scrobbles": 0, "scrobble": True}
+ self.config.register_global(version=1)
+ self.config.register_user(**defaults)
+ self.config.register_guild(crowns={})
+ self.session = aiohttp.ClientSession(
+ headers={
+ "User-Agent": "Mozilla/5.0 (X11; Arch Linux; Linux x86_64; rv:66.0) Gecko/20100101 Firefox/66.0"
+ },
+ )
+ self.token = None
+ self.wc = None
+ self.login_token = None
+ self.wordcloud_create()
+ self.data_loc = bundled_data_path(self)
+ self.chart_data = {}
+ self.chart_data_loop = self.bot.loop.create_task(self.chart_clear_loop())
+
+ def format_help_for_context(self, ctx):
+ pre_processed = super().format_help_for_context(ctx)
+ return f"{pre_processed}\n\nCog Version: {self.__version__}"
+
+ async def red_delete_data_for_user(self, *, requester, user_id):
+ await self.config.user_from_id(user_id).clear()
+
+ async def chart_clear_loop(self):
+ await self.bot.wait_until_ready()
+ while True:
+ self.chart_data = {}
+ await asyncio.sleep(1800)
+
+ async def initialize(self):
+ token = await self.bot.get_shared_api_tokens("lastfm")
+ self.token = token.get("appid")
+ self.secret = token.get("secret")
+ self.login_token = token.get("logintoken")
+ await self.migrate_config()
+
+ async def migrate_config(self):
+ if await self.config.version() == 1:
+ a = {}
+ conf = await self.config.all_guilds()
+ for guild in conf:
+ a[guild] = {"crowns": {}}
+ for artist in conf[guild]["crowns"]:
+ a[guild]["crowns"][artist.lower()] = conf[guild]["crowns"][artist]
+ group = self.config._get_base_group(self.config.GUILD)
+ async with group.all() as new_data:
+ for guild in a:
+ new_data[guild] = a[guild]
+ await self.config.version.set(2)
+
+ @commands.Cog.listener(name="on_red_api_tokens_update")
+ async def listener_update_class_tokens(self, service_name, api_tokens):
+ if service_name == "lastfm":
+ self.token = api_tokens.get("appid")
+ self.secret = api_tokens.get("secret")
+ self.login_token = api_tokens.get("logintoken")
+
+ def cog_unload(self):
+ self.bot.loop.create_task(self.session.close())
+ if self.chart_data_loop:
+ self.chart_data_loop.cancel()
+
+ @commands.is_owner()
+ @commands.command(name="lastfmset", aliases=["fmset"])
+ async def command_lastfmset(self, ctx):
+ """Instructions on how to set the api key."""
+ message = (
+ "1. Visit the [LastFM](https://www.last.fm/api/) website and click on 'Get an API Account'.\n"
+ "2. Fill in the application. Once completed do not exit the page. - "
+ "Copy all information on the page and save it.\n"
+ f"3. Enter the api key via `{ctx.clean_prefix}set api lastfm appid `\n"
+ f"4. Enter the api secret via `{ctx.clean_prefix}set api lastfm secret `\n"
+ f"--------\n"
+ f"Some commands that use webscraping may require a login token.\n"
+ f"1. Visit [LastFM](https://www.last.fm) site and login.\n"
+ f"2. Open your browser's developer tools and go to the Storage tab.\n"
+ f"3. Find the cookie named `sessionid` and copy the value.\n"
+ f"4. Enter the api secret via `{ctx.clean_prefix}set api lastfm logintoken `\n"
+ )
+ await ctx.maybe_send_embed(message)
+
+ @commands.command(name="crowns")
+ @commands.check(tokencheck)
+ @commands.guild_only()
+ async def command_crowns(self, ctx, user: discord.Member = None):
+ """Check yourself or another users crowns."""
+ user = user or ctx.author
+ crowns = await self.config.guild(ctx.guild).crowns()
+ crownartists = []
+ for key in crowns:
+ if crowns[key]["user"] == user.id:
+ crownartists.append((key, crowns[key]["playcount"]))
+ if crownartists is None:
+ return await ctx.send(
+ "You haven't acquired any crowns yet! "
+ f"Use the `{ctx.clean_prefix}whoknows` command to claim crowns \N{CROWN}"
+ )
+
+ rows = []
+ for artist, playcount in sorted(crownartists, key=itemgetter(1), reverse=True):
+ rows.append(f"**{artist}** with **{playcount}** {self.format_plays(playcount)}")
+
+ content = discord.Embed(
+ title=f"Artist crowns for {user.name} — Total {len(crownartists)} crowns",
+ color=user.color,
+ )
+ content.set_footer(text="Playcounts are updated on the whoknows command.")
+ if not rows:
+ return await ctx.send("You do not have any crowns.")
+ pages = await self.create_pages(content, rows)
+ if len(pages) > 1:
+ await menu(ctx, pages, DEFAULT_CONTROLS)
+ else:
+ await ctx.send(embed=pages[0])
+
+ @command_fm.command(
+ name="artist", usage="[timeframe] "
+ )
+ async def command_artist(self, ctx, timeframe, datatype, *, artistname=""):
+ """Your top tracks or albums for specific artist.
+
+ Usage:
+ [p]fm artist [timeframe] toptracks
+ [p]fm artist [timeframe] topalbums
+ [p]fm artist [timeframe] overview """
+ conf = await self.config.user(ctx.author).all()
+ username = conf["lastfm_username"]
+
+ period, _ = self.get_period(timeframe)
+ if period in [None, "today"]:
+ artistname = " ".join([datatype, artistname]).strip()
+ datatype = timeframe
+ period = "overall"
+
+ artistname = self.remove_mentions(artistname)
+
+ if artistname == "":
+ return await ctx.send("Missing artist name!")
+
+ if datatype in ["toptracks", "tt", "tracks", "track"]:
+ datatype = "tracks"
+
+ elif datatype in ["topalbums", "talb", "albums", "album"]:
+ datatype = "albums"
+
+ elif datatype in ["overview", "stats", "ov"]:
+ return await self.artist_overview(ctx, period, artistname, username)
+
+ else:
+ return await ctx.send_help()
+
+ artist, data = await self.artist_top(ctx, period, artistname, datatype, username)
+ if artist is None or not data:
+ artistname = escape(artistname)
+ if period == "overall":
+ return await ctx.send(f"You have never listened to **{artistname}**!")
+ else:
+ return await ctx.send(
+ f"You have not listened to **{artistname}** in the past {period}s!"
+ )
+
+ total = 0
+ rows = []
+ for i, (name, playcount) in enumerate(data, start=1):
+ rows.append(
+ f"`#{i:2}` **{playcount}** {self.format_plays(playcount)} — **{escape(name)}**"
+ )
+ total += playcount
+
+ artistname = urllib.parse.quote_plus(artistname)
+ content = discord.Embed(color=await ctx.embed_color())
+ content.set_thumbnail(url=artist["image_url"])
+ content.set_author(
+ name=f"{ctx.author.display_name} — "
+ + (f"{self.humanized_period(period)} " if period != "overall" else "")
+ + f"Top {datatype} by {artist['formatted_name']}",
+ icon_url=ctx.author.display_avatar.url,
+ url=f"https://last.fm/user/{username}/library/music/{artistname}/"
+ f"+{datatype}?date_preset={self.period_http_format(period)}",
+ )
+ content.set_footer(
+ text=f"Total {total} {self.format_plays(total)} across {len(rows)} {datatype}"
+ )
+
+ pages = await self.create_pages(content, rows)
+ if len(pages) > 1:
+ await menu(ctx, pages[:15], DEFAULT_CONTROLS)
+ else:
+ await ctx.send(embed=pages[0])
+
+ @command_fm.command(name="last")
+ async def command_last(self, ctx):
+ """
+ Your weekly listening overview.
+ """
+ conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(conf)
+ await self.listening_report(ctx, "week", conf["lastfm_username"])
+
+ @command_fm.command(name="lyrics", aliases=["lyr"])
+ async def command_lyrics(self, ctx, *, track: str = None):
+ """Currently playing song or most recent song."""
+ if track is None:
+ conf = await self.config.user(ctx.author).all()
+ self.check_if_logged_in(conf)
+ track, artist, albumname, image_url = await self.get_current_track(
+ ctx, conf["lastfm_username"]
+ )
+
+ title = (
+ f"**{escape(artist, formatting=True)}** — ***{escape(track, formatting=True)} ***"
+ )
+
+ results, songtitle = await self.lyrics_musixmatch(f"{artist} {track}")
+ if results is None:
+ return await ctx.send(f'No lyrics for "{artist} {track}" found.')
+ embeds = []
+ results = list(pagify(results, page_length=2048))
+ for i, page in enumerate(results, 1):
+ content = discord.Embed(
+ color=await ctx.embed_color(),
+ description=page,
+ title=title,
+ )
+ content.set_thumbnail(url=image_url)
+ if len(results) > 1:
+ content.set_footer(text=f"Page {i}/{len(results)}")
+
+ embeds.append(content)
+ if len(embeds) > 1:
+ await menu(ctx, embeds, DEFAULT_CONTROLS)
+ else:
+ await ctx.send(embed=embeds[0])
+ else:
+
+ results, songtitle = await self.lyrics_musixmatch(track)
+ if results is None:
+ return await ctx.send(f'No lyrics for "{track}" found.')
+ embeds = []
+ results = list(pagify(results, page_length=2048))
+ for i, page in enumerate(results, 1):
+ content = discord.Embed(
+ color=await ctx.embed_color(),
+ title=f"***{escape(songtitle, formatting=True)} ***",
+ description=page,
+ )
+ if len(results) > 1:
+ content.set_footer(text=f"Page {i}/{len(results)}")
+ embeds.append(content)
+ if len(embeds) > 1:
+ await menu(ctx, embeds, DEFAULT_CONTROLS)
+ else:
+ await ctx.send(embed=embeds[0])
+
+ @command_fm.command(name="streak")
+ async def command_streak(self, ctx, user: discord.User = None):
+ """
+ View how many times you've listened to something in a row
+
+ Only the most 200 recent plays are tracked
+ """
+ if not user:
+ user = ctx.author
+ conf = await self.config.user(user).all()
+ self.check_if_logged_in(conf, user == ctx.author)
+ data = await self.api_request(
+ ctx,
+ {"user": conf["lastfm_username"], "method": "user.getrecenttracks", "limit": 200},
+ )
+ tracks = data["recenttracks"]["track"]
+ if not tracks or not isinstance(tracks, list):
+ return await ctx.send("You have not listened to anything yet!")
+ track_streak = [tracks[0]["name"], 1, True]
+ artist_streak = [tracks[0]["artist"]["#text"], 1, True]
+ album_streak = [tracks[0]["album"]["#text"], 1, True]
+ ignore = True
+ for x in tracks:
+ if ignore:
+ ignore = False
+ continue
+ if track_streak[2]:
+ if x["name"] == track_streak[0]:
+ track_streak[1] += 1
+ else:
+ track_streak[2] = False
+ if artist_streak[2]:
+ if x["artist"]["#text"] == artist_streak[0]:
+ artist_streak[1] += 1
+ else:
+ artist_streak[2] = False
+ if album_streak[2]:
+ if x["album"]["#text"] == album_streak[0]:
+ album_streak[1] += 1
+ else:
+ album_streak[2] = False
+
+ if not track_streak[2] and not artist_streak[2] and not album_streak[2]:
+ break
+
+ if track_streak[1] == 1 and artist_streak[1] == 1 and album_streak[1] == 1:
+ return await ctx.send("You have not listened to anything in a row.")
+ embed = discord.Embed(color=await ctx.embed_color(), title=f"{user.name}'s streaks")
+ embed.set_thumbnail(url=tracks[0]["image"][3]["#text"])
+ if track_streak[1] > 1:
+ embed.add_field(
+ name="Track", value=f"{track_streak[1]} times in a row \n({track_streak[0][:50]})"
+ )
+ if artist_streak[1] > 1:
+ embed.add_field(
+ name="Artist",
+ value=f"{artist_streak[1]} times in a row \n({artist_streak[0][:50]})",
+ )
+ if album_streak[1] > 1:
+ embed.add_field(
+ name="Album", value=f"{album_streak[1]} times in a row \n({album_streak[0][:50]})"
+ )
+
+ await ctx.send(embed=embed)
+
+ async def cog_command_error(self, ctx, error):
+ if hasattr(error, "original"):
+ if isinstance(error.original, SilentDeAuthorizedError):
+ return
+ if isinstance(error.original, LastFMError):
+ await ctx.send(str(error.original))
+ return
+ await ctx.bot.on_command_error(ctx, error, unhandled_by_cog=True)
diff --git a/lastfm/love.py b/lastfm/love.py
new file mode 100644
index 0000000..d1283ae
--- /dev/null
+++ b/lastfm/love.py
@@ -0,0 +1,145 @@
+import discord
+from redbot.core.utils.menus import DEFAULT_CONTROLS, menu
+
+from .abc import MixinMeta
+from .exceptions import *
+from .fmmixin import FMMixin
+
+command_fm = FMMixin.command_fm
+command_fm_server = FMMixin.command_fm_server
+
+
+class LoveMixin(MixinMeta):
+ """Love Commands"""
+
+ async def love_or_unlove_song(self, track, artist, love, key):
+ params = {
+ "api_key": self.token,
+ "artist": artist,
+ "sk": key,
+ "track": track,
+ }
+ if love:
+ params["method"] = "track.love"
+ else:
+ params["method"] = "track.unlove"
+ data = await self.api_post(params=params)
+ return data
+
+ @command_fm.command(name="love", usage="