refactor: refactor for code smells too many branches and too many statements

Author: douglasessousa

bot/exts/filtering/_filter_context.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import typing
 from collections.abc import Callable, Coroutine, Iterable
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass, field, replace as dataclass_replace
 from enum import Enum, auto
 
 import discord
@@ -25,60 +27,124 @@ class Event(Enum):
 
 
 @dataclass
-class FilterContext:
-    """A dataclass containing the information that should be filtered, and output information of the filtering."""
-
-    # 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.
+class FilterSource:
+    """The source/sender metadata for a filtering context."""
+
+    event: Event
+    author: User | Member | None
+    channel: TextChannel | VoiceChannel | StageChannel | Thread | DMChannel | None
+    message: Message | None
     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)
-    potential_phish: dict[FilterList, set[str]] = field(default_factory=dict)
-    # Additional actions to perform
+
+
+@dataclass
+class FilterContent:
+    """The content being filtered."""
+
+    content: str | Iterable
+    embeds: list[Embed] = field(default_factory=list)
+    attachments: list[discord.Attachment | FileAttachment] = field(default_factory=list)
+
+
+@dataclass
+class FilterNotifications:
+    """DM and alert content produced by filtering."""
+
+    dm_content: str = ""
+    dm_embed: str = ""
+    send_alert: bool = False
+    alert_content: str = ""
+    alert_embeds: list[Embed] = field(default_factory=list)
+    notification_domain: str = ""
+    action_descriptions: list[str] = field(default_factory=list)
+
+
+@dataclass
+class FilterActions:
+    """Side effects and deletion metadata produced by filtering."""
+
     additional_actions: list[Callable[[FilterContext], Coroutine]] = field(default_factory=list)
-    related_messages: set[Message] = field(default_factory=set)  # Deletion will include these.
+    messages_deletion: bool = False
+    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
+
+
+@dataclass
+class FilterResults:
+    """Filter match results and tracking data."""
+
+    matches: list[str] = field(default_factory=list)
+    filter_info: dict[Filter, str] = field(default_factory=dict)
+    blocked_exts: set[str] = field(default_factory=set)
+    potential_phish: dict[FilterList, set[str]] = field(default_factory=dict)
+
 
-    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
+class FilterContext:
+    """A context object containing the information that should be filtered, and output information of the filtering.
+
+    Attributes are delegated to sub-objects for organization:
+    - ``source``: event, author, channel, message, before_message, message_cache
+    - ``content``: content, embeds, attachments
+    - ``notifications``: dm_content, dm_embed, send_alert, alert_content, alert_embeds, notification_domain, action_descriptions
+    - ``actions``: additional_actions, messages_deletion, related_messages, related_channels, uploaded_attachments, upload_deletion_logs
+    - ``results``: matches, filter_info, blocked_exts, potential_phish
+    """
+
+    def __init__(self, source, content, notifications=None, actions=None, results=None):
+        self._source = source
+        self._content = content
+        self._notifications = notifications or FilterNotifications()
+        self._actions = actions or FilterActions()
+        self._results = results or FilterResults()
+        self.in_guild = source.channel is None or source.channel.guild is not None
+
+    def __getattr__(self, name):
+        for obj in (self._source, self._content, self._notifications, self._actions, self._results):
+            if hasattr(obj, name):
+                return getattr(obj, name)
+        raise AttributeError(f"'{type(self).__name__}' has no attribute '{name}'")
+
+    def __setattr__(self, name, value):
+        if name.startswith('_') or name == 'in_guild':
+            object.__setattr__(self, name, value)
+            return
+        for obj in (self._source, self._content, self._notifications, self._actions, self._results):
+            if hasattr(obj, name):
+                setattr(obj, name, value)
+                return
+        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
-        )
+        source = FilterSource(event, message.author, message.channel, message, before, cache)
+        content = FilterContent(message.content, message.embeds, message.attachments)
+        return cls(source, content)
 
     def replace(self, **changes) -> FilterContext:
         """Return a new context object assigning new values to the specified fields."""
-        return replace(self, **changes)
+        sub_objects = {
+            '_source': self._source,
+            '_content': self._content,
+            '_notifications': self._notifications,
+            '_actions': self._actions,
+            '_results': self._results,
+        }
+        sub_changes = {}
+        for key, value in changes.items():
+            for attr_name, obj in sub_objects.items():
+                if hasattr(obj, key):
+                    sub_changes.setdefault(attr_name, {})[key] = value
+                    break
+        return FilterContext(
+            source=dataclass_replace(self._source, **sub_changes.get('_source', {})),
+            content=dataclass_replace(self._content, **sub_changes.get('_content', {})),
+            notifications=dataclass_replace(self._notifications, **sub_changes.get('_notifications', {})),
+            actions=dataclass_replace(self._actions, **sub_changes.get('_actions', {})),
+            results=dataclass_replace(self._results, **sub_changes.get('_results', {})),
+        )

bot/exts/filtering/_filter_lists/antispam.py

@@ -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 FilterContent, FilterContext, FilterSource
 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
@@ -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(FilterSource(ctx.event, ctx.author, ctx.channel, ctx.message), FilterContent(ctx.content))
         all_descriptions_counts = Counter(reduce(
             add, (other_ctx.action_descriptions for other_ctx in other_contexts), ctx.action_descriptions
         ))

bot/exts/filtering/_filters/filter.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any
 
 import arrow
@@ -9,6 +10,13 @@
 from bot.exts.filtering._utils import FieldRequiring
 
 
+@dataclass
+class FilterTimestamps:
+    """Timestamps for when a filter was created and last updated."""
+    created_at: arrow.Arrow
+    updated_at: arrow.Arrow
+
+
 class Filter(FieldRequiring):
     """
     A class representing a filter.
@@ -27,8 +35,10 @@ 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"])
@@ -50,6 +60,11 @@ def overrides(self) -> tuple[dict[str, Any], dict[str, Any]]:
 
         return settings, filter_settings
 
+    @property
+    def last_updated(self) -> arrow.Arrow:
+        """The most recent time this filter was created or updated."""
+        return max(self.timestamps.created_at, self.timestamps.updated_at)
+
     @abstractmethod
     async def triggered_on(self, ctx: FilterContext) -> bool:
         """Search for the filter's content within a given context."""

bot/exts/filtering/_loaded_types.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from bot.exts.filtering._filters.filter import Filter
+    from bot.exts.filtering._settings_types.settings_entry import SettingsEntry
+
+
+@dataclass
+class LoadedTypes:
+    """Container for loaded type metadata used across the filtering UI."""
+
+    filters: dict[str, type["Filter"]]
+    settings: dict[str, tuple[str, "SettingsEntry", type]]
+    filter_settings: dict[str, dict[str, tuple[str, "SettingsEntry", type]]]

bot/exts/filtering/_ui/filter.py

@@ -10,6 +10,7 @@
 
 from bot.exts.filtering._filter_lists.filter_list import FilterList, ListType
 from bot.exts.filtering._filters.filter import Filter
+from bot.exts.filtering._loaded_types import LoadedTypes
 from bot.exts.filtering._ui.ui import (
     COMPONENT_TIMEOUT,
     CustomCallbackSelect,
@@ -124,8 +125,7 @@ def __init__(
         description: str | None,
         settings_overrides: dict,
         filter_settings_overrides: dict,
-        loaded_settings: dict,
-        loaded_filter_settings: dict,
+        loaded: LoadedTypes,
         author: User,
         embed: Embed,
         confirm_callback: Callable
@@ -138,8 +138,7 @@ def __init__(
         self.description = description
         self.settings_overrides = settings_overrides
         self.filter_settings_overrides = filter_settings_overrides
-        self.loaded_settings = loaded_settings
-        self.loaded_filter_settings = loaded_filter_settings
+        self.loaded = loaded
         self.embed = embed
         self.confirm_callback = confirm_callback
 
@@ -148,10 +147,10 @@ def __init__(
         )
         populate_embed_from_dict(embed, all_settings_repr_dict)
 
-        self.type_per_setting_name = {setting: info[2] for setting, info in loaded_settings.items()}
+        self.type_per_setting_name = {setting: info[2] for setting, info in loaded.settings.items()}
         self.type_per_setting_name.update({
             f"{filter_type.name}/{name}": type_
-            for name, (_, _, type_) in loaded_filter_settings.get(filter_type.name, {}).items()
+            for name, (_, _, type_) in loaded.filter_settings.get(filter_type.name, {}).items()
         })
 
         add_select = CustomCallbackSelect(
@@ -381,8 +380,7 @@ def copy(self) -> FilterEditView:
             self.description,
             self.settings_overrides,
             self.filter_settings_overrides,
-            self.loaded_settings,
-            self.loaded_filter_settings,
+            self.loaded,
             self.author,
             self.embed,
             self.confirm_callback
@@ -395,10 +393,10 @@ def _parse_filter_list_setting(
     settings: dict,
     filter_list: FilterList,
     list_type: ListType,
-    loaded_settings: dict,
+    loaded: LoadedTypes,
 ) -> None:
     """Parse and validate a filter list setting, updating `settings` in place."""
-    type_ = loaded_settings[setting][2]
+    type_ = loaded.settings[setting][2]
     try:
         parsed_value = parse_value(value, type_)
         if not repr_equals(parsed_value, filter_list[list_type].default(setting)):
@@ -412,17 +410,17 @@ def _parse_filter_setting(
     value: str,
     filter_settings: dict,
     filter_type: type[Filter],
-    loaded_filter_settings: dict,
+    loaded: LoadedTypes,
 ) -> None:
     """Parse and validate a filter-specific setting, updating `filter_settings` in place."""
     filter_name, filter_setting_name = setting.split("/", maxsplit=1)
     if filter_name.lower() != filter_type.name.lower():
         raise BadArgument(
             f"A setting for a {filter_name!r} filter was provided, but the filter name is {filter_type.name!r}"
         )
-    if filter_setting_name not in loaded_filter_settings[filter_type.name]:
+    if filter_setting_name not in loaded.filter_settings[filter_type.name]:
         raise BadArgument(f"{setting!r} is not a recognized setting.")
-    type_ = loaded_filter_settings[filter_type.name][filter_setting_name][2]
+    type_ = loaded.filter_settings[filter_type.name][filter_setting_name][2]
     try:
         parsed_value = parse_value(value, type_)
         if not repr_equals(parsed_value, getattr(filter_type.extra_fields_type(), filter_setting_name)):
@@ -436,19 +434,18 @@ def _parse_settings(
     filter_list: FilterList,
     list_type: ListType,
     filter_type: type[Filter],
-    loaded_settings: dict,
-    loaded_filter_settings: dict,
+    loaded: LoadedTypes,
 ) -> tuple[dict, dict]:
     """Parse and validate all settings, returning (list_settings, filter_settings)."""
     settings = {}
     filter_settings = {}
     for setting, value in raw_settings.items():
-        if setting in loaded_settings:
-            _parse_filter_list_setting(setting, value, settings, filter_list, list_type, loaded_settings)
+        if setting in loaded.settings:
+            _parse_filter_list_setting(setting, value, settings, filter_list, list_type, loaded)
         elif "/" not in setting:
             raise BadArgument(f"{setting!r} is not a recognized setting.")
         else:
-            _parse_filter_setting(setting, value, filter_settings, filter_type, loaded_filter_settings)
+            _parse_filter_setting(setting, value, filter_settings, filter_type, loaded)
     return settings, filter_settings
 
 
@@ -472,8 +469,7 @@ def description_and_settings_converter(
     filter_list: FilterList,
     list_type: ListType,
     filter_type: type[Filter],
-    loaded_settings: dict,
-    loaded_filter_settings: dict,
+    loaded: LoadedTypes,
     input_data: str
 ) -> tuple[str, dict[str, Any], dict[str, Any]]:
     """Parse a string representing a possible description and setting overrides, and validate the setting names."""
@@ -492,7 +488,7 @@ def description_and_settings_converter(
     template = raw_settings.pop("--template", None)
 
     settings, filter_settings = _parse_settings(
-        raw_settings, filter_list, list_type, filter_type, loaded_settings, loaded_filter_settings
+        raw_settings, filter_list, list_type, filter_type, loaded
     )
 
     if template is not None: