Compare commits
No commits in common. "v0.2.1" and "master" have entirely different histories.
@ -1,4 +1,4 @@
|
||||
[bumpversion]
|
||||
current_version = 0.2.1
|
||||
current_version = 0.2.0
|
||||
commit = True
|
||||
tag = True
|
||||
|
||||
@ -1,12 +1,4 @@
|
||||
|
||||
**v0.2.1**
|
||||
|
||||
- Enhancement: view filters command
|
||||
- Enhancement: more control over view commands
|
||||
- Enhancement: show active state of viewed subscriptions
|
||||
- Fix: bug where certain RSS feeds without a build date would break the task
|
||||
- Fix: TypeError when an RSS Item has no title, description or publish date.
|
||||
|
||||
**v0.2.0**
|
||||
|
||||
- Fix: Fetch channels if not found in bot cache (error fix)
|
||||
|
||||
@ -157,7 +157,7 @@ class API:
|
||||
|
||||
log.debug("getting tracked content")
|
||||
|
||||
return await self._get_many(self.API_ENDPOINT + "tracked-content/", filters)
|
||||
return await self._get_many(self.API_ENDPOINT + f"tracked-content/", filters)
|
||||
|
||||
async def get_filter(self, filter_id: int) -> dict:
|
||||
"""
|
||||
@ -167,10 +167,3 @@ class API:
|
||||
log.debug("getting a filter")
|
||||
|
||||
return await self._get_one(f"{self.API_ENDPOINT}filter/{filter_id}")
|
||||
|
||||
async def get_filters(self, **filters) -> tuple[list[dict], int]:
|
||||
"""
|
||||
Get many instances of Filter.
|
||||
"""
|
||||
|
||||
return await self._get_many(self.API_ENDPOINT + "filter/", filters)
|
||||
|
||||
@ -12,11 +12,11 @@ import validators
|
||||
from feedparser import FeedParserDict, parse
|
||||
from discord.ext import commands
|
||||
from discord import Interaction, TextChannel, Embed, Colour
|
||||
from discord.app_commands import Choice, choices, Group, autocomplete, rename, command
|
||||
from discord.app_commands import Choice, Group, autocomplete, rename, command
|
||||
from discord.errors import Forbidden
|
||||
|
||||
from api import API
|
||||
from feed import Subscription, TrackedContent, ContentFilter
|
||||
from feed import Subscription, TrackedContent
|
||||
from utils import (
|
||||
Followup,
|
||||
PaginationView,
|
||||
@ -78,12 +78,6 @@ async def validate_rss_source(nickname: str, url: str) -> Tuple[str | None, Feed
|
||||
|
||||
return None, feed
|
||||
|
||||
tri_choices = [
|
||||
Choice(name="Yes", value=2),
|
||||
Choice(name="No (default)", value=1),
|
||||
Choice(name="All", value=0),
|
||||
]
|
||||
|
||||
|
||||
class CommandsCog(commands.Cog):
|
||||
"""
|
||||
@ -115,13 +109,14 @@ class CommandsCog(commands.Cog):
|
||||
|
||||
def formatdata(index, item):
|
||||
item = Subscription.from_dict(item)
|
||||
|
||||
|
||||
channels = f"{item.channels_count}{' channels' if item.channels_count != 1 else ' channel'}"
|
||||
filters = f"{len(item.filters)}{' filters' if len(item.filters) != 1 else ' filter'}"
|
||||
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}/)"
|
||||
activeness = "✅ `enabled`" if item.active else "🚫 `disabled`"
|
||||
links = f"[RSS Link]({item.url}) · [API Link]({API.API_EXTERNAL_ENDPOINT}subscription/{item.id}/)"
|
||||
|
||||
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"{channels}, {filters}\n"
|
||||
description += f"{notes}\n" if notes else ""
|
||||
description += links
|
||||
|
||||
key = f"{index}. {item.name}"
|
||||
@ -150,104 +145,32 @@ class CommandsCog(commands.Cog):
|
||||
await pagination.send()
|
||||
|
||||
@view_group.command(name="tracked-content")
|
||||
@choices(blocked=tri_choices)
|
||||
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
|
||||
async def cmd_list_tracked(self, inter: Interaction, search: str = ""):
|
||||
"""List Tracked Content from this server, or a given sub"""
|
||||
|
||||
await inter.response.defer()
|
||||
|
||||
# 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.
|
||||
if isinstance(blocked, Choice):
|
||||
blocked = blocked.value
|
||||
|
||||
def formatdata(index, item):
|
||||
item = TrackedContent.from_dict(item)
|
||||
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}/)"
|
||||
delivery_state = "✅ Delivered" if not item.blocked else "🚫 Blocked"
|
||||
links = f"[Content Link]({item.url}) · [Message Link](https://discord.com/channels/{sub.guild_id}/{item.channel_id}/{item.message_id}/)"
|
||||
description = f"Subscription: {sub.name}\n{links}"
|
||||
|
||||
description = f"🆔 `{item.id}`\n"
|
||||
description += f"{delivery_state}\n" if blocked == 0 else ""
|
||||
description += f"➡️ *{sub.name}*\n{links}"
|
||||
|
||||
key = f"{index}. {item.title}"
|
||||
key = f"{item.id}. {item.title}"
|
||||
return key, description
|
||||
|
||||
def determine_blocked():
|
||||
match blocked:
|
||||
case 0: return ""
|
||||
case 1: return "false"
|
||||
case 2: return "true"
|
||||
case _: return ""
|
||||
|
||||
async def getdata(page: int, pagesize: int):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
api = API(self.bot.api_token, session)
|
||||
is_blocked = determine_blocked()
|
||||
return await api.get_tracked_content(
|
||||
subscription__guild_id=inter.guild_id,
|
||||
blocked=is_blocked,
|
||||
page=page,
|
||||
page_size=pagesize,
|
||||
search=search,
|
||||
)
|
||||
|
||||
embed = Followup(f"Tracked Content in {inter.guild.name}").info()._embed
|
||||
pagination = PaginationView(
|
||||
self.bot,
|
||||
inter=inter,
|
||||
embed=embed,
|
||||
getdata=getdata,
|
||||
formatdata=formatdata,
|
||||
pagesize=10,
|
||||
initpage=1
|
||||
)
|
||||
await pagination.send()
|
||||
|
||||
@view_group.command(name="filters")
|
||||
async def cmd_list_filters(self, inter: Interaction, search: str = ""):
|
||||
"""List Filters from this server."""
|
||||
|
||||
await inter.response.defer()
|
||||
|
||||
def formatdata(index, item):
|
||||
item = ContentFilter.from_dict(item)
|
||||
|
||||
matching_algorithm = get_algorithm_name(item.matching_algorithm)
|
||||
whitelist = "Whitelist" if item.is_whitelist else "Blacklist"
|
||||
sensitivity = "Case insensitive" if item.is_insensitive else "Case sensitive"
|
||||
|
||||
description = f"🆔 `{item.id}`\n"
|
||||
description += f"🔄 `{matching_algorithm}`\n🟰 `{item.match}`\n"
|
||||
description += f"✅ `{whitelist}` 🔠 `{sensitivity}`\n"
|
||||
description += f"[API Link]({API.API_EXTERNAL_ENDPOINT}filter/{item.id}/)"
|
||||
|
||||
key = f"{index}. {item.name}"
|
||||
return key, description
|
||||
|
||||
def get_algorithm_name(matching_algorithm: int):
|
||||
match matching_algorithm:
|
||||
case 0: return "None"
|
||||
case 1: return "Any word"
|
||||
case 2: return "All words"
|
||||
case 3: return "Exact match"
|
||||
case 4: return "Regex match"
|
||||
case 5: return "Fuzzy match"
|
||||
case _: return "unknown"
|
||||
|
||||
async def getdata(page, pagesize):
|
||||
async with aiohttp.ClientSession() as session:
|
||||
api = API(self.bot.api_token, session)
|
||||
return await api.get_filters(
|
||||
guild_id=inter.guild_id,
|
||||
page=page,
|
||||
page_size=pagesize,
|
||||
search=search
|
||||
)
|
||||
)
|
||||
|
||||
embed = Followup(f"Filters in {inter.guild.name}").info()._embed
|
||||
embed = Followup(f"Tracked Content in {inter.guild.name}").info()._embed
|
||||
pagination = PaginationView(
|
||||
self.bot,
|
||||
inter=inter,
|
||||
|
||||
25
src/feed.py
25
src/feed.py
@ -199,7 +199,7 @@ class RSSFeed:
|
||||
description: str
|
||||
link: str
|
||||
lang: str
|
||||
last_build_date: datetime | None
|
||||
last_build_date: datetime
|
||||
image_href: str
|
||||
items: list[RSSItem] = None
|
||||
|
||||
@ -240,8 +240,7 @@ class RSSFeed:
|
||||
language = pf.feed.get('language', None)
|
||||
|
||||
last_build_date = pf.feed.get('updated_parsed', None)
|
||||
if last_build_date:
|
||||
last_build_date = datetime(*last_build_date[0:-2])
|
||||
last_build_date = datetime(*last_build_date[0:-2] if last_build_date else None)
|
||||
|
||||
image_href = pf.feed.get("image", {}).get("href")
|
||||
|
||||
@ -251,7 +250,7 @@ class RSSFeed:
|
||||
item = RSSItem.from_parsed_entry(entry)
|
||||
feed.add_item(item)
|
||||
|
||||
feed.items.reverse() # order so that older items are processed first
|
||||
feed.items.reverse()
|
||||
return feed
|
||||
|
||||
|
||||
@ -358,21 +357,3 @@ class TrackedContent(DjangoDataModel):
|
||||
|
||||
item["creation_datetime"] = datetime.strptime(item["creation_datetime"], "%Y-%m-%dT%H:%M:%S.%f%z")
|
||||
return item
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ContentFilter(DjangoDataModel):
|
||||
|
||||
id: int
|
||||
name: str
|
||||
matching_algorithm: int
|
||||
match: str
|
||||
is_insensitive: bool
|
||||
is_whitelist: bool
|
||||
guild_id: int
|
||||
|
||||
@staticmethod
|
||||
def parser(item: dict) -> dict:
|
||||
|
||||
item["guild_id"] = int(item["guild_id"]) # stored as str due to a django/sqlite bug, convert back to int
|
||||
return item
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user