corbz/PYRSS-Bot
corbz/PYRSS-Bot
1
0
Fork 0
Code Issues 14 Pull Requests Actions Releases 2 Activity

temp trim commands
Some checks are pending
Build and Push Docker Image / build (push) Waiting to run
Details

Browse Source
This commit is contained in:
Corban-Lee Jones 2024-10-31 13:10:46 +00:00
parent eb97dca5c6
commit cf8fb34a29
1 changed files with 222 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

444
src/extensions/cmds.py
View File

@ -7,82 +7,82 @@ import logging
from typing import Tuple from typing import Tuple
from datetime import datetime from datetime import datetime
import aiohttp # import aiohttp
import validators # import validators
from feedparser import FeedParserDict, parse from feedparser import FeedParserDict, parse
from discord.ext import commands from discord.ext import commands
from discord import Interaction, TextChannel, Embed, Colour from discord import Interaction, TextChannel, Embed, Colour
from discord.app_commands import Choice, choices, Group, autocomplete, rename, command from discord.app_commands import Choice, choices, Group, autocomplete, rename, command
from discord.errors import Forbidden from discord.errors import Forbidden
from api import API # from api import API
from feed import Subscription, TrackedContent, ContentFilter # from feed import Subscription, TrackedContent, ContentFilter
from utils import ( # from utils import (
Followup, # Followup,
PaginationView, # PaginationView,
get_rss_data, # get_rss_data,
) # )
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
rss_list_sort_choices = [ # rss_list_sort_choices = [
Choice(name="Nickname", value=0), # Choice(name="Nickname", value=0),
Choice(name="Date Added", value=1) # Choice(name="Date Added", value=1)
] # ]
channels_list_sort_choices=[ # channels_list_sort_choices=[
Choice(name="Feed Nickname", value=0), # Choice(name="Feed Nickname", value=0),
Choice(name="Channel ID", value=1), # Choice(name="Channel ID", value=1),
Choice(name="Date Added", value=2) # Choice(name="Date Added", value=2)
] # ]
# TODO SECURITY: a potential attack is that the user submits an rss feed then changes the # # TODO SECURITY: a potential attack is that the user submits an rss feed then changes the
# target resource. Run a period task to check this. # # target resource. Run a period task to check this.
async def validate_rss_source(nickname: str, url: str) -> Tuple[str | None, FeedParserDict | None]: # async def validate_rss_source(nickname: str, url: str) -> Tuple[str | None, FeedParserDict | None]:
"""Validate a provided RSS source. # """Validate a provided RSS source.
Parameters # Parameters
---------- # ----------
nickname : str # nickname : str
Nickname of the source. Must not contain URL. # Nickname of the source. Must not contain URL.
url : str # url : str
URL of the source. Must be URL with valid status code and be an RSS feed. # URL of the source. Must be URL with valid status code and be an RSS feed.
Returns # Returns
------- # -------
str or None # str or None
String invalid message if invalid, NoneType if valid. # String invalid message if invalid, NoneType if valid.
FeedParserDict or None # FeedParserDict or None
The feed parsed from the given URL or None if invalid. # The feed parsed from the given URL or None if invalid.
""" # """
# Ensure the URL is valid # # Ensure the URL is valid
if not validators.url(url): # if not validators.url(url):
return f"The URL you have entered is malformed or invalid:\n`{url=}`", None # return f"The URL you have entered is malformed or invalid:\n`{url=}`", None
# Check the nickname is not a URL # # Check the nickname is not a URL
if validators.url(nickname): # if validators.url(nickname):
return "It looks like the nickname you have entered is a URL.\n" \ # return "It looks like the nickname you have entered is a URL.\n" \
f"For security reasons, this is not allowed.\n`{nickname=}`", None # f"For security reasons, this is not allowed.\n`{nickname=}`", None
feed_data, status_code = await get_rss_data(url) # feed_data, status_code = await get_rss_data(url)
# Check the URL status code is valid # # Check the URL status code is valid
if status_code != 200: # if status_code != 200:
return f"The URL provided returned an invalid status code:\n{url=}, {status_code=}", None # return f"The URL provided returned an invalid status code:\n{url=}, {status_code=}", None
# Check the contents is actually an RSS feed. # # Check the contents is actually an RSS feed.
feed = parse(feed_data) # feed = parse(feed_data)
if not feed.version: # if not feed.version:
return f"The provided URL '{url}' does not seem to be a valid RSS feed.", None # return f"The provided URL '{url}' does not seem to be a valid RSS feed.", None
return None, feed # return None, feed
tri_choices = [ # tri_choices = [
Choice(name="Yes", value=2), # Choice(name="Yes", value=2),
Choice(name="No (default)", value=1), # Choice(name="No (default)", value=1),
Choice(name="All", value=0), # Choice(name="All", value=0),
] # ]
class CommandsCog(commands.Cog): class CommandsCog(commands.Cog):
@ -100,206 +100,206 @@ class CommandsCog(commands.Cog):
log.info("%s cog is ready", self.__class__.__name__) log.info("%s cog is ready", self.__class__.__name__)
# Group for commands about viewing data # # Group for commands about viewing data
view_group = Group( # view_group = Group(
name="view", # name="view",
description="View data.", # description="View data.",
guild_only=True # guild_only=True
) # )
@view_group.command(name="subscriptions") # @view_group.command(name="subscriptions")
async def cmd_list_subs(self, inter: Interaction, search: str = ""): # async def cmd_list_subs(self, inter: Interaction, search: str = ""):
"""List Subscriptions from this server.""" # """List Subscriptions from this server."""
await inter.response.defer() # await inter.response.defer()
def formatdata(index, item): # def formatdata(index, item):
item = Subscription.from_dict(item) # item = Subscription.from_dict(item)
notes = item.extra_notes[:25] + "..." if len(item.extra_notes) > 28 else item.extra_notes # notes = item.extra_notes[:25] + "..." if len(item.extra_notes) > 28 else item.extra_notes
links = f"[RSS Link]({item.url}) • [API Link]({API.API_EXTERNAL_ENDPOINT}subscription/{item.id}/)" # links = f"[RSS Link]({item.url}) • [API Link]({API.API_EXTERNAL_ENDPOINT}subscription/{item.id}/)"
activeness = "✅ `enabled`" if item.active else "🚫 `disabled`" # activeness = "✅ `enabled`" if item.active else "🚫 `disabled`"
description = f"🆔 `{item.id}`\n{activeness}\n#️⃣ `{item.channels_count}` 🔽 `{len(item.filters)}`\n" # description = f"🆔 `{item.id}`\n{activeness}\n#️⃣ `{item.channels_count}` 🔽 `{len(item.filters)}`\n"
description = f"{notes}\n" + description if notes else description # description = f"{notes}\n" + description if notes else description
description += links # description += links
key = f"{index}. {item.name}" # key = f"{index}. {item.name}"
return key, description # key, value pair # return key, description # key, value pair
async def getdata(page: int, pagesize: int): # async def getdata(page: int, pagesize: int):
async with aiohttp.ClientSession() as session: # async with aiohttp.ClientSession() as session:
api = API(self.bot.api_token, session) # api = API(self.bot.api_token, session)
return await api.get_subscriptions( # return await api.get_subscriptions(
guild_id=inter.guild.id, # guild_id=inter.guild.id,
page=page, # page=page,
page_size=pagesize, # page_size=pagesize,
search=search # search=search
) # )
embed = Followup(f"Subscriptions in {inter.guild.name}").info()._embed # embed = Followup(f"Subscriptions in {inter.guild.name}").info()._embed
pagination = PaginationView( # pagination = PaginationView(
self.bot, # self.bot,
inter=inter, # inter=inter,
embed=embed, # embed=embed,
getdata=getdata, # getdata=getdata,
formatdata=formatdata, # formatdata=formatdata,
pagesize=10, # pagesize=10,
initpage=1 # initpage=1
) # )
await pagination.send() # await pagination.send()
@view_group.command(name="tracked-content") # @view_group.command(name="tracked-content")
@choices(blocked=tri_choices) # @choices(blocked=tri_choices)
async def cmd_list_tracked(self, inter: Interaction, search: str = "", blocked: Choice[int] = 1): # async def cmd_list_tracked(self, inter: Interaction, search: str = "", blocked: Choice[int] = 1):
"""List Tracked Content from this server""" # TODO: , or a given sub # """List Tracked Content from this server""" # TODO: , or a given sub
await inter.response.defer() # await inter.response.defer()
# If the user picks an option it's an instance of `Choice` otherwise `str` # # If the user picks an option it's an instance of `Choice` otherwise `str`
# Can't figure a way to select a default choices, so blame discordpy for this mess. # # Can't figure a way to select a default choices, so blame discordpy for this mess.
if isinstance(blocked, Choice): # if isinstance(blocked, Choice):
blocked = blocked.value # blocked = blocked.value
def formatdata(index, item): # def formatdata(index, item):
item = TrackedContent.from_dict(item) # item = TrackedContent.from_dict(item)
sub = Subscription.from_dict(item.subscription) # sub = Subscription.from_dict(item.subscription)
links = f"[Content Link]({item.url}) · [Message Link](https://discord.com/channels/{sub.guild_id}/{item.channel_id}/{item.message_id}/) · [API Link]({API.API_EXTERNAL_ENDPOINT}tracked-content/{item.id}/)" # links = f"[Content Link]({item.url}) · [Message Link](https://discord.com/channels/{sub.guild_id}/{item.channel_id}/{item.message_id}/) · [API Link]({API.API_EXTERNAL_ENDPOINT}tracked-content/{item.id}/)"
delivery_state = "✅ Delivered" if not item.blocked else "🚫 Blocked" # delivery_state = "✅ Delivered" if not item.blocked else "🚫 Blocked"
description = f"🆔 `{item.id}`\n" # description = f"🆔 `{item.id}`\n"
description += f"{delivery_state}\n" if blocked == 0 else "" # description += f"{delivery_state}\n" if blocked == 0 else ""
description += f"➡️ *{sub.name}*\n{links}" # description += f"➡️ *{sub.name}*\n{links}"
key = f"{index}. {item.title}" # key = f"{index}. {item.title}"
return key, description # return key, description
def determine_blocked(): # def determine_blocked():
match blocked: # match blocked:
case 0: return "" # case 0: return ""
case 1: return "false" # case 1: return "false"
case 2: return "true" # case 2: return "true"
case _: return "" # case _: return ""
async def getdata(page: int, pagesize: int): # async def getdata(page: int, pagesize: int):
async with aiohttp.ClientSession() as session: # async with aiohttp.ClientSession() as session:
api = API(self.bot.api_token, session) # api = API(self.bot.api_token, session)
is_blocked = determine_blocked() # is_blocked = determine_blocked()
return await api.get_tracked_content( # return await api.get_tracked_content(
subscription__guild_id=inter.guild_id, # subscription__guild_id=inter.guild_id,
blocked=is_blocked, # blocked=is_blocked,
page=page, # page=page,
page_size=pagesize, # page_size=pagesize,
search=search, # search=search,
) # )
embed = Followup(f"Tracked Content in {inter.guild.name}").info()._embed # embed = Followup(f"Tracked Content in {inter.guild.name}").info()._embed
pagination = PaginationView( # pagination = PaginationView(
self.bot, # self.bot,
inter=inter, # inter=inter,
embed=embed, # embed=embed,
getdata=getdata, # getdata=getdata,
formatdata=formatdata, # formatdata=formatdata,
pagesize=10, # pagesize=10,
initpage=1 # initpage=1
) # )
await pagination.send() # await pagination.send()
@view_group.command(name="filters") # @view_group.command(name="filters")
async def cmd_list_filters(self, inter: Interaction, search: str = ""): # async def cmd_list_filters(self, inter: Interaction, search: str = ""):
"""List Filters from this server.""" # """List Filters from this server."""
await inter.response.defer() # await inter.response.defer()
def formatdata(index, item): # def formatdata(index, item):
item = ContentFilter.from_dict(item) # item = ContentFilter.from_dict(item)
matching_algorithm = get_algorithm_name(item.matching_algorithm) # matching_algorithm = get_algorithm_name(item.matching_algorithm)
whitelist = "Whitelist" if item.is_whitelist else "Blacklist" # whitelist = "Whitelist" if item.is_whitelist else "Blacklist"
sensitivity = "Case insensitive" if item.is_insensitive else "Case sensitive" # sensitivity = "Case insensitive" if item.is_insensitive else "Case sensitive"
description = f"🆔 `{item.id}`\n" # description = f"🆔 `{item.id}`\n"
description += f"🔄 `{matching_algorithm}`\n🟰 `{item.match}`\n" # description += f"🔄 `{matching_algorithm}`\n🟰 `{item.match}`\n"
description += f"✅ `{whitelist}` 🔠 `{sensitivity}`\n" # description += f"✅ `{whitelist}` 🔠 `{sensitivity}`\n"
description += f"[API Link]({API.API_EXTERNAL_ENDPOINT}filter/{item.id}/)" # description += f"[API Link]({API.API_EXTERNAL_ENDPOINT}filter/{item.id}/)"
key = f"{index}. {item.name}" # key = f"{index}. {item.name}"
return key, description # return key, description
def get_algorithm_name(matching_algorithm: int): # def get_algorithm_name(matching_algorithm: int):
match matching_algorithm: # match matching_algorithm:
case 0: return "None" # case 0: return "None"
case 1: return "Any word" # case 1: return "Any word"
case 2: return "All words" # case 2: return "All words"
case 3: return "Exact match" # case 3: return "Exact match"
case 4: return "Regex match" # case 4: return "Regex match"
case 5: return "Fuzzy match" # case 5: return "Fuzzy match"
case _: return "unknown" # case _: return "unknown"
async def getdata(page, pagesize): # async def getdata(page, pagesize):
async with aiohttp.ClientSession() as session: # async with aiohttp.ClientSession() as session:
api = API(self.bot.api_token, session) # api = API(self.bot.api_token, session)
return await api.get_filters( # return await api.get_filters(
guild_id=inter.guild_id, # guild_id=inter.guild_id,
page=page, # page=page,
page_size=pagesize, # page_size=pagesize,
search=search # search=search
) # )
embed = Followup(f"Filters in {inter.guild.name}").info()._embed # embed = Followup(f"Filters in {inter.guild.name}").info()._embed
pagination = PaginationView( # pagination = PaginationView(
self.bot, # self.bot,
inter=inter, # inter=inter,
embed=embed, # embed=embed,
getdata=getdata, # getdata=getdata,
formatdata=formatdata, # formatdata=formatdata,
pagesize=10, # pagesize=10,
initpage=1 # initpage=1
) # )
await pagination.send() # await pagination.send()
# Group for test related commands # # Group for test related commands
test_group = Group( # test_group = Group(
name="test", # name="test",
description="Commands to test Bot functionality.", # description="Commands to test Bot functionality.",
guild_only=True # guild_only=True
) # )
@test_group.command(name="channel-permissions") # @test_group.command(name="channel-permissions")
async def cmd_test_channel_perms(self, inter: Interaction): # async def cmd_test_channel_perms(self, inter: Interaction):
"""Test that the current channel's permissions allow for PYRSS to operate in it.""" # """Test that the current channel's permissions allow for PYRSS to operate in it."""
try: # try:
test_message = await inter.channel.send(content="... testing permissions ...") # test_message = await inter.channel.send(content="... testing permissions ...")
await self.test_channel_perms(inter.channel) # await self.test_channel_perms(inter.channel)
except Exception as error: # except Exception as error:
await inter.response.send_message(content=f"Failed: {error}") # await inter.response.send_message(content=f"Failed: {error}")
return # return
await test_message.delete() # await test_message.delete()
await inter.response.send_message(content="Success") # await inter.response.send_message(content="Success")
async def test_channel_perms(self, channel: TextChannel): # async def test_channel_perms(self, channel: TextChannel):
# Test generic message and delete # # Test generic message and delete
msg = await channel.send(content="test message") # msg = await channel.send(content="test message")
await msg.delete() # await msg.delete()
# Test detailed embed # # Test detailed embed
embed = Embed( # embed = Embed(
title="test title", # title="test title",
description="test description", # description="test description",
colour=Colour.random(), # colour=Colour.random(),
timestamp=datetime.now(), # timestamp=datetime.now(),
url="https://google.com" # url="https://google.com"
) # )
embed.set_author(name="test author") # embed.set_author(name="test author")
embed.set_footer(text="test footer") # embed.set_footer(text="test footer")
embed.set_thumbnail(url="https://www.google.com/images/branding/googlelogo/2x/googlelogo_light_color_272x92dp.png") # embed.set_thumbnail(url="https://www.google.com/images/branding/googlelogo/2x/googlelogo_light_color_272x92dp.png")
embed.set_image(url="https://www.google.com/images/branding/googlelogo/2x/googlelogo_light_color_272x92dp.png") # embed.set_image(url="https://www.google.com/images/branding/googlelogo/2x/googlelogo_light_color_272x92dp.png")
embed_msg = await channel.send(embed=embed) # embed_msg = await channel.send(embed=embed)
await embed_msg.delete() # await embed_msg.delete()
async def setup(bot): async def setup(bot):