Refactor membership queries

cogs/make_member.py

@@ -1,19 +1,16 @@
 """Contains cog classes for any make_member interactions."""
 
 import logging
-import re
 from typing import TYPE_CHECKING
 
-import aiohttp
-import bs4
 import discord
-from bs4 import BeautifulSoup
 from django.core.exceptions import ValidationError
 
 from config import settings
 from db.core.models import GroupMadeMember
 from exceptions import ApplicantRoleDoesNotExistError, GuestRoleDoesNotExistError
-from utils import GLOBAL_SSL_CONTEXT, CommandChecks, TeXBotBaseCog
+from utils import CommandChecks, TeXBotBaseCog
+from utils.msl import fetch_community_group_members_count, is_id_a_community_group_member
 
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
@@ -116,6 +113,20 @@ async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: st
         member_role: discord.Role = await self.bot.member_role
         interaction_member: discord.Member = await ctx.bot.get_main_guild_member(ctx.user)
 
+        INVALID_GUILD_MEMBER_ID: Final[str] = (
+            f"{group_member_id!r} is not a valid {self.bot.group_member_id_type} ID."
+        )
+
+        try:
+            group_member_id_int: int = int(group_member_id)
+        except ValueError:
+            await self.command_send_error(ctx=ctx, message=INVALID_GUILD_MEMBER_ID)
+            return
+
+        if group_member_id_int < 1000000 or group_member_id_int > 99999999:
+            await self.command_send_error(ctx=ctx, message=INVALID_GUILD_MEMBER_ID)
+            return
+
         await ctx.defer(ephemeral=True)
         async with ctx.typing():
             if member_role in interaction_member.roles:
@@ -128,19 +139,9 @@ async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: st
                 )
                 return
 
-            if not re.fullmatch(r"\A\d{7}\Z", group_member_id):
-                await self.command_send_error(
-                    ctx,
-                    message=(
-                        f"{group_member_id!r} is not a valid "
-                        f"{self.bot.group_member_id_type} ID."
-                    ),
-                )
-                return
-
             if await GroupMadeMember.objects.filter(
                 hashed_group_member_id=GroupMadeMember.hash_group_member_id(
-                    group_member_id, self.bot.group_member_id_type
+                    group_member_id_int, self.bot.group_member_id_type
                 )
             ).aexists():
                 await ctx.followup.send(
@@ -154,58 +155,9 @@ async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: st
                 )
                 return
 
-            guild_member_ids: set[str] = set()
-
-            async with (
-                aiohttp.ClientSession(
-                    headers=REQUEST_HEADERS, cookies=REQUEST_COOKIES
-                ) as http_session,
-                http_session.get(
-                    url=GROUPED_MEMBERS_URL, ssl=GLOBAL_SSL_CONTEXT
-                ) as http_response,
-            ):
-                response_html: str = await http_response.text()
-
-            MEMBER_HTML_TABLE_IDS: Final[frozenset[str]] = frozenset(
-                {
-                    "ctl00_Main_rptGroups_ctl05_gvMemberships",
-                    "ctl00_Main_rptGroups_ctl03_gvMemberships",
-                    "ctl00_ctl00_Main_AdminPageContent_rptGroups_ctl03_gvMemberships",
-                    "ctl00_ctl00_Main_AdminPageContent_rptGroups_ctl05_gvMemberships",
-                }
-            )
-            table_id: str
-            for table_id in MEMBER_HTML_TABLE_IDS:
-                parsed_html: bs4.Tag | bs4.NavigableString | None = BeautifulSoup(
-                    response_html, "html.parser"
-                ).find("table", {"id": table_id})
-
-                if parsed_html is None or isinstance(parsed_html, bs4.NavigableString):
-                    continue
-
-                guild_member_ids.update(
-                    row.contents[2].text
-                    for row in parsed_html.find_all("tr", {"class": ["msl_row", "msl_altrow"]})
-                )
-
-            guild_member_ids.discard("")
-            guild_member_ids.discard("\n")
-            guild_member_ids.discard(" ")
-
-            if not guild_member_ids:
+            if not await is_id_a_community_group_member(student_id=group_member_id_int):
                 await self.command_send_error(
-                    ctx,
-                    error_code="E1041",
-                    logging_message=OSError(
-                        "The guild member IDs could not be retrieved from "
-                        "the MEMBERS_LIST_URL."
-                    ),
-                )
-                return
-
-            if group_member_id not in guild_member_ids:
-                await self.command_send_error(
-                    ctx,
+                    ctx=ctx,
                     message=(
                         f"You must be a member of {self.bot.group_full_name} "
                         "to use this command.\n"
@@ -276,53 +228,9 @@ async def member_count(self, ctx: "TeXBotApplicationContext") -> None:  # type:
         await ctx.defer(ephemeral=False)
 
         async with ctx.typing():
-            async with (
-                aiohttp.ClientSession(
-                    headers=REQUEST_HEADERS, cookies=REQUEST_COOKIES
-                ) as http_session,
-                http_session.get(
-                    url=BASE_MEMBERS_URL, ssl=GLOBAL_SSL_CONTEXT
-                ) as http_response,
-            ):
-                response_html: str = await http_response.text()
-
-            member_list_div: bs4.Tag | bs4.NavigableString | None = BeautifulSoup(
-                response_html, "html.parser"
-            ).find("div", {"class": "memberlistcol"})
-
-            if member_list_div is None or isinstance(member_list_div, bs4.NavigableString):
-                await self.command_send_error(
-                    ctx,
-                    error_code="E1041",
-                    logging_message=OSError(
-                        "The member count could not be retrieved from the MEMBERS_LIST_URL."
-                    ),
-                )
-                return
-
-            if "showing 100 of" in member_list_div.text.lower():
-                member_count: str = member_list_div.text.split(" ")[3]
-                await ctx.followup.send(
-                    content=f"{self.bot.group_full_name} has {member_count} members! :tada:"
-                )
-                return
-
-            member_table: bs4.Tag | bs4.NavigableString | None = BeautifulSoup(
-                response_html, "html.parser"
-            ).find("table", {"id": "ctl00_ctl00_Main_AdminPageContent_gvMembers"})
-
-            if member_table is None or isinstance(member_table, bs4.NavigableString):
-                await self.command_send_error(
-                    ctx,
-                    error_code="E1041",
-                    logging_message=OSError(
-                        "The member count could not be retrieved from the MEMBERS_LIST_URL."
-                    ),
-                )
-                return
-
             await ctx.followup.send(
-                content=f"{self.bot.group_full_name} has {
-                    len(member_table.find_all('tr', {'class': ['msl_row', 'msl_altrow']}))
-                } members! :tada:"
+                content=(
+                    f"{self.bot.group_full_name} has "
+                    f"{await fetch_community_group_members_count()} members! :tada:"
+                )
             )

exceptions/__init__.py

@@ -24,6 +24,7 @@
     MessagesJSONFileMissingKeyError,
     MessagesJSONFileValueError,
 )
+from .msl import MSLMembershipError
 from .strike import NoAuditLogsStrikeTrackingError, StrikeTrackingError
 
 if TYPE_CHECKING:
@@ -44,6 +45,7 @@
     "InvalidActionDescriptionError",
     "InvalidActionTargetError",
     "InvalidMessagesJSONFileError",
+    "MSLMembershipError",
     "MemberRoleDoesNotExistError",
     "MessagesJSONFileMissingKeyError",
     "MessagesJSONFileValueError",

exceptions/msl.py

@@ -0,0 +1,26 @@
+"""Custom exception classes raised when errors occur during use of MSL features."""
+
+from typing import TYPE_CHECKING, override
+
+from typed_classproperties import classproperty
+
+from .base import BaseTeXBotError
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+
+__all__: "Sequence[str]" = ("MSLMembershipError",)
+
+
+class MSLMembershipError(BaseTeXBotError, RuntimeError):
+    """
+    Exception class to raise when any error occurs while checking MSL membership.
+
+    If this error occurs, it is likely that MSL features will not work correctly.
+    """
+
+    @classproperty
+    @override
+    def DEFAULT_MESSAGE(cls) -> str:
+        return "An error occurred while trying to fetch membership data from MSL."

utils/msl/__init__.py

@@ -0,0 +1,18 @@
+"""MSL utility classes & functions provided for use across the whole of the project."""
+
+from typing import TYPE_CHECKING
+
+from .memberships import (
+    fetch_community_group_members_count,
+    fetch_community_group_members_list,
+    is_id_a_community_group_member,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+__all__: "Sequence[str]" = (
+    "fetch_community_group_members_count",
+    "fetch_community_group_members_list",
+    "is_id_a_community_group_member",
+)

utils/msl/memberships.py

@@ -0,0 +1,153 @@
+"""Module for checking membership status."""
+
+import contextlib
+import logging
+from typing import TYPE_CHECKING
+
+import aiohttp
+import bs4
+from bs4 import BeautifulSoup
+
+from config import settings
+from exceptions import MSLMembershipError
+from utils import GLOBAL_SSL_CONTEXT
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+    from http.cookies import Morsel
+    from logging import Logger
+    from typing import Final
+
+
+__all__: "Sequence[str]" = (
+    "fetch_community_group_members_count",
+    "fetch_community_group_members_list",
+    "is_id_a_community_group_member",
+)
+
+
+BASE_SU_PLATFORM_WEB_HEADERS: "Final[Mapping[str, str]]" = {
+    "Cache-Control": "no-cache",
+    "Pragma": "no-cache",
+    "Expires": "0",
+}
+
+
+BASE_SU_PLATFORM_WEB_COOKIES: "Final[Mapping[str, str]]" = {
+    ".ASPXAUTH": settings["SU_PLATFORM_ACCESS_COOKIE"],
+}
+
+
+MEMBERS_LIST_URL: "Final[str]" = f"https://guildofstudents.com/organisation/memberlist/{settings['ORGANISATION_ID']}/?sort=groups"
+
+
+_membership_list_cache: set[int] = set()
+
+
+logger: "Final[Logger]" = logging.getLogger("TeX-Bot")
+
+
+async def fetch_msl_context(url: str) -> tuple[dict[str, str], dict[str, str]]:
+    """Get the required context headers, data and cookies to make a request to MSL."""
+    http_session: aiohttp.ClientSession = aiohttp.ClientSession(
+        headers=BASE_SU_PLATFORM_WEB_HEADERS,
+        cookies=BASE_SU_PLATFORM_WEB_COOKIES,
+    )
+    data_fields: dict[str, str] = {}
+    cookies: dict[str, str] = {}
+    async with http_session, http_session.get(url=url, ssl=GLOBAL_SSL_CONTEXT) as field_data:
+        data_response = BeautifulSoup(
+            markup=await field_data.text(),
+            features="html.parser",
+        )
+
+        for field in data_response.find_all(name="input"):
+            if field.get("name") and field.get("value"):
+                data_fields[field.get("name")] = field.get("value")
+
+        for cookie in field_data.cookies:
+            cookie_morsel: Morsel[str] | None = field_data.cookies.get(cookie)
+            if cookie_morsel is not None:
+                cookies[cookie] = cookie_morsel.value
+        cookies[".ASPXAUTH"] = settings["MEMBERS_LIST_AUTH_SESSION_COOKIE"]
+
+    return data_fields, cookies
+
+
+async def fetch_community_group_members_list() -> set[int]:
+    """
+    Make a web request to fetch your community group's full membership list.
+
+    Returns a set of IDs.
+    """
+    async with (
+        aiohttp.ClientSession(
+            headers=BASE_SU_PLATFORM_WEB_HEADERS, cookies=BASE_SU_PLATFORM_WEB_COOKIES
+        ) as http_session,
+        http_session.get(url=MEMBERS_LIST_URL, ssl=GLOBAL_SSL_CONTEXT) as http_response,
+    ):
+        response_html: str = await http_response.text()
+
+    parsed_html: BeautifulSoup = BeautifulSoup(markup=response_html, features="html.parser")
+
+    member_ids: set[int] = set()
+
+    table_id: str
+    for table_id in (
+        "ctl00_ctl00_Main_AdminPageContent_rptGroups_ctl03_gvMemberships",
+        "ctl00_ctl00_Main_AdminPageContent_rptGroups_ctl05_gvMemberships",
+    ):
+        filtered_table: bs4.Tag | bs4.NavigableString | None = parsed_html.find(
+            name="table", attrs={"id": table_id}
+        )
+
+        if filtered_table is None:
+            MEMBER_TABLE_ERROR: str = (
+                f"Membership table with ID {table_id} could not be found."
+            )
+            logger.warning(MEMBER_TABLE_ERROR)
+            logger.debug(response_html)
+            continue
+
+        if isinstance(filtered_table, bs4.NavigableString):
+            MEMBER_TABLE_FORMAT_ERROR: str = (
+                f"Membership table with ID {table_id} was found but is in the wrong format."
+            )
+            logger.warning(MEMBER_TABLE_FORMAT_ERROR)
+            logger.debug(filtered_table)
+            raise MSLMembershipError(message=MEMBER_TABLE_FORMAT_ERROR)
+
+        with contextlib.suppress(IndexError):
+            rows: list[bs4.Tag] = filtered_table.find_all(name="tr")[1:]
+            for member in rows:
+                with contextlib.suppress(ValueError):
+                    member_ids.add(int(member.find_all(name="td")[1].text.strip()))
+
+    if not member_ids:  # NOTE: this should never be possible, because to fetch the page you need to have admin access, which requires being a member.
+        NO_MEMBERS_ERROR: str = "No members were found in either membership table."
+        logger.warning(NO_MEMBERS_ERROR)
+        logger.debug(response_html)
+        raise MSLMembershipError(message=NO_MEMBERS_ERROR)
+
+    _membership_list_cache.clear()
+    _membership_list_cache.update(member_ids)
+
+    return _membership_list_cache
+
+
+async def is_id_a_community_group_member(student_id: int) -> bool:
+    """Check if the given ID is a member of your community group."""
+    if student_id in _membership_list_cache:
+        return True
+
+    logger.debug(
+        "ID %s not found in community group membership list cache; Fetching updated list.",
+        student_id,
+    )
+
+    return student_id in await fetch_community_group_members_list()
+
+
+async def fetch_community_group_members_count() -> int:
+    """Return the total number of members in your community group."""
+    return len(await fetch_community_group_members_list())