Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -125,3 +125,17 @@ TEST-**.xml
# Mac OS .DS_Store, which is a file that stores custom attributes of its containing folder
.DS_Store
*.env*

# # Métricas
/metrics-before-radon
/metrics-before-pylint
/metrics-after-pylint
/metrics-before-codecarbon
/metrics-before-pytest

extract_metrics_before_radon.py
extract_metrics_before_pylint.py
extract_metrics_after_pylint.py
extract_score_before_pylint.py
extract_metrics_before_codecarbon.py
extract_metrics_before_pytest.py
148 changes: 109 additions & 39 deletions bot/exts/filtering/_filter_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,60 +25,130 @@ class Event(Enum):


@dataclass
class FilterContext:
"""A dataclass containing the information that should be filtered, and output information of the filtering."""
class FilterInput: # pylint: disable=too-many-instance-attributes
"""Input data for filtering: event details and message content."""

# Input context
event: Event # The type of event
author: User | Member | None # Who triggered the event
channel: TextChannel | VoiceChannel | StageChannel | Thread | DMChannel | None # The channel involved
content: str | Iterable # What actually needs filtering. The Iterable type depends on the filter list.
message: Message | None # The message involved
embeds: list[Embed] = field(default_factory=list) # Any embeds involved
attachments: list[discord.Attachment | FileAttachment] = field(default_factory=list) # Any attachments sent.
event: Event
author: User | Member | None
channel: TextChannel | VoiceChannel | StageChannel | Thread | DMChannel | None
content: str | Iterable
message: Message | None
embeds: list[Embed] = field(default_factory=list)
attachments: list[discord.Attachment | FileAttachment] = field(default_factory=list)
before_message: Message | None = None
message_cache: MessageCache | None = None
# Output context
dm_content: str = "" # The content to DM the invoker
dm_embed: str = "" # The embed description to DM the invoker
send_alert: bool = False # Whether to send an alert for the moderators
alert_content: str = "" # The content of the alert
alert_embeds: list[Embed] = field(default_factory=list) # Any embeds to add to the alert
action_descriptions: list[str] = field(default_factory=list) # What actions were taken
matches: list[str] = field(default_factory=list) # What exactly was found
notification_domain: str = "" # A domain to send the user for context
filter_info: dict[Filter, str] = field(default_factory=dict) # Additional info from a filter.
messages_deletion: bool = False # Whether the messages were deleted. Can't upload deletion log otherwise.
blocked_exts: set[str] = field(default_factory=set) # Any extensions blocked (used for snekbox)


@dataclass
class FilterOutput: # pylint: disable=too-many-instance-attributes
"""Output data produced by filtering: alerts, actions, and results."""

dm_content: str = ""
dm_embed: str = ""
send_alert: bool = False
alert_content: str = ""
alert_embeds: list[Embed] = field(default_factory=list)
action_descriptions: list[str] = field(default_factory=list)
matches: list[str] = field(default_factory=list)
notification_domain: str = ""
filter_info: dict[Filter, str] = field(default_factory=dict)
messages_deletion: bool = False
blocked_exts: set[str] = field(default_factory=set)
potential_phish: dict[FilterList, set[str]] = field(default_factory=dict)
# Additional actions to perform


_FILTER_CONTEXT_DIRECT_FIELDS = frozenset({
'input', 'output', 'additional_actions', 'related_messages',
'related_channels', 'uploaded_attachments', 'upload_deletion_logs',
})


@dataclass
class FilterContext:
"""A dataclass containing the information that should be filtered, and output information of the filtering."""

input: FilterInput
output: FilterOutput
additional_actions: list[Callable[[FilterContext], Coroutine]] = field(default_factory=list)
related_messages: set[Message] = field(default_factory=set) # Deletion will include these.
related_messages: set[Message] = field(default_factory=set)
related_channels: set[TextChannel | Thread | DMChannel] = field(default_factory=set)
uploaded_attachments: dict[int, list[str]] = field(default_factory=dict) # Message ID to attachment URLs.
upload_deletion_logs: bool = True # Whether it's allowed to upload deletion logs.
uploaded_attachments: dict[int, list[str]] = field(default_factory=dict)
upload_deletion_logs: bool = True

@property
def in_guild(self) -> bool:
"""Whether the context is from a guild channel (not a DM)."""
return self.input.channel is None or self.input.channel.guild is not None

def __getattr__(self, name):
try:
input_obj = object.__getattribute__(self, 'input')
if hasattr(input_obj, name):
return getattr(input_obj, name)
except AttributeError:
pass
try:
output_obj = object.__getattribute__(self, 'output')
if hasattr(output_obj, name):
return getattr(output_obj, name)
except AttributeError:
pass
raise AttributeError(f"'FilterContext' has no attribute '{name}'")

def __post_init__(self):
# If it's in the context of a DM channel, self.channel won't be None, but self.channel.guild will.
self.in_guild = self.channel is None or self.channel.guild is not None
def __setattr__(self, name, value):
if name in _FILTER_CONTEXT_DIRECT_FIELDS:
object.__setattr__(self, name, value)
return
try:
input_obj = object.__getattribute__(self, 'input')
if hasattr(input_obj, name):
setattr(input_obj, name, value)
return
except AttributeError:
pass
try:
output_obj = object.__getattribute__(self, 'output')
if hasattr(output_obj, name):
setattr(output_obj, name, value)
return
except AttributeError:
pass
object.__setattr__(self, name, value)

@classmethod
def from_message(
cls, event: Event, message: Message, before: Message | None = None, cache: MessageCache | None = None
) -> FilterContext:
"""Create a filtering context from the attributes of a message."""
return cls(
event,
message.author,
message.channel,
message.content,
message,
message.embeds,
message.attachments,
before,
cache
FilterInput(
event,
message.author,
message.channel,
message.content,
message,
message.embeds,
message.attachments,
before,
cache
),
FilterOutput()
)

def replace(self, **changes) -> FilterContext:
"""Return a new context object assigning new values to the specified fields."""
return replace(self, **changes)
input_fields = FilterInput.__dataclass_fields__
output_fields = FilterOutput.__dataclass_fields__
input_changes = {}
output_changes = {}
context_changes = {}
for k, v in changes.items():
if k in input_fields:
input_changes[k] = v
elif k in output_fields:
output_changes[k] = v
else:
context_changes[k] = v
new_input = replace(self.input, **input_changes) if input_changes else self.input
new_output = replace(self.output, **output_changes) if output_changes else self.output
return FilterContext(new_input, new_output, **context_changes)
4 changes: 2 additions & 2 deletions bot/exts/filtering/_filter_lists/antispam.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from pydis_core.utils import scheduling
from pydis_core.utils.logging import get_logger

from bot.exts.filtering._filter_context import FilterContext
from bot.exts.filtering._filter_context import FilterContext, FilterInput, FilterOutput
from bot.exts.filtering._filter_lists.filter_list import ListType, SubscribingAtomicList, UniquesListBase
from bot.exts.filtering._filters.antispam import antispam_filter_types
from bot.exts.filtering._filters.filter import Filter, UniqueFilter
Expand Down Expand Up @@ -158,7 +158,7 @@ async def send_alert(self, antispam_list: AntispamList) -> None:
return

ctx, *other_contexts = self.contexts
new_ctx = FilterContext(ctx.event, ctx.author, ctx.channel, ctx.content, ctx.message)
new_ctx = FilterContext(FilterInput(ctx.event, ctx.author, ctx.channel, ctx.content, ctx.message), FilterOutput())
all_descriptions_counts = Counter(reduce(
add, (other_ctx.action_descriptions for other_ctx in other_contexts), ctx.action_descriptions
))
Expand Down
27 changes: 24 additions & 3 deletions bot/exts/filtering/_filters/filter.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,16 @@
from bot.exts.filtering._filter_context import Event, FilterContext
from bot.exts.filtering._settings import Defaults, create_settings
from bot.exts.filtering._utils import FieldRequiring
from dataclasses import dataclass

import arrow


@dataclass
class FilterTimestamps:
"""Timestamps for when a filter was created and last updated."""
created_at: arrow.Arrow
updated_at: arrow.Arrow


class Filter(FieldRequiring):
Expand All @@ -23,12 +33,14 @@ class Filter(FieldRequiring):
# If a subclass uses extra fields, it should assign the pydantic model type to this variable.
extra_fields_type = None

def __init__(self, filter_data: dict, defaults: Defaults | None = None):
def __init__(self, filter_data: dict, defaults: Defaults | None=None):
self.id = filter_data["id"]
self.content = filter_data["content"]
self.description = filter_data["description"]
self.created_at = arrow.get(filter_data["created_at"])
self.updated_at = arrow.get(filter_data["updated_at"])
self.timestamps = FilterTimestamps(
created_at=arrow.get(filter_data["created_at"]),
updated_at=arrow.get(filter_data["updated_at"])
)
self.actions, self.validations = create_settings(filter_data["settings"], defaults=defaults)
if self.extra_fields_type:
self.extra_fields = self.extra_fields_type.model_validate(filter_data["additional_settings"])
Expand Down Expand Up @@ -75,6 +87,15 @@ async def process_input(cls, content: str, description: str) -> tuple[str, str]:
A BadArgument should be raised if the content can't be used.
"""
return content, description


@property
def created_at(self) -> arrow.Arrow:
return self.timestamps.created_at

@property
def updated_at(self) -> arrow.Arrow:
return self.timestamps.updated_at

def __str__(self) -> str:
"""A string representation of the filter."""
Expand Down
Loading
Loading