Add msl methods

cogs/make_member.py

@@ -5,10 +5,7 @@
 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
@@ -27,6 +24,8 @@
 
     from utils import TeXBotApplicationContext
 
+from utils.msl import get_membership_count, is_student_id_member
+
 __all__: "Sequence[str]" = ("MakeMemberCommandCog", "MemberCountCommandCog")
 
 logger: "Final[Logger]" = logging.getLogger("TeX-Bot")
@@ -118,7 +117,7 @@ class MakeMemberCommandCog(TeXBotBaseCog):
         parameter_name="group_member_id",
     )
     @CommandChecks.check_interaction_user_in_main_guild
-    async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: str) -> None:  # type: ignore[misc]  # noqa: PLR0915
+    async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: str) -> None:  # type: ignore[misc]
         """
         Definition & callback response of the "make_member" command.
 
@@ -160,11 +159,9 @@ async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: st
                 )
             ).aexists()
             if GROUP_MEMBER_ID_IS_ALREADY_USED:
-                # noinspection PyUnusedLocal
                 committee_mention: str = "committee"
                 with contextlib.suppress(CommitteeRoleDoesNotExistError):
                     committee_mention = (await self.bot.committee_role).mention
-
                 await ctx.followup.send(
                     content=(
                         ":information_source: No changes made. This student ID has already "
@@ -175,92 +172,32 @@ async def make_member(self, ctx: "TeXBotApplicationContext", group_member_id: st
                 )
                 return
 
-            guild_member_ids: set[str] = set()
-
-            http_session: aiohttp.ClientSession = aiohttp.ClientSession(
-                headers=REQUEST_HEADERS,
-                cookies=REQUEST_COOKIES,
+        if not await is_student_id_member(student_id=group_member_id):
+            await self.command_send_error(
+                ctx=ctx,
+                message=(
+                    f"You must be a member of {self.bot.group_full_name} "
+                    "to use this command.\n"
+                    f"The provided {_GROUP_MEMBER_ID_ARGUMENT_NAME} must match "
+                    f"the {self.bot.group_member_id_type} ID "
+                    f"that you purchased your {self.bot.group_short_name} membership with."
+                ),
             )
-            async with http_session, http_session.get(GROUPED_MEMBRS_URL) 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:
-                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,
-                    message=(
-                        f"You must be a member of {self.bot.group_full_name} "
-                        "to use this command.\n"
-                        f"The provided {_GROUP_MEMBER_ID_ARGUMENT_NAME} must match "
-                        f"the {self.bot.group_member_id_type} ID "
-                        f"that you purchased your {self.bot.group_short_name} membership with."
-                    ),
+        try:
+            await GroupMadeMember.objects.acreate(group_member_id=group_member_id)  # type: ignore[misc]
+        except ValidationError as create_group_made_member_error:
+            error_is_already_exists: bool = (
+                "hashed_group_member_id" in create_group_made_member_error.message_dict
+                and any(
+                    "already exists" in error
+                    for error in create_group_made_member_error.message_dict[
+                        "hashed_group_member_id"
+                    ]
                 )
-                return
-
-            # NOTE: The "Member" role must be added to the user **before** the "Guest" role to ensure that the welcome message does not include the suggestion to purchase membership
-            await interaction_member.add_roles(
-                member_role,
-                reason='TeX Bot slash-command: "/makemember"',
             )
-
-            try:
-                await GroupMadeMember.objects.acreate(group_member_id=group_member_id)  # type: ignore[misc]
-            except ValidationError as create_group_made_member_error:
-                error_is_already_exists: bool = (
-                    "hashed_group_member_id" in create_group_made_member_error.message_dict
-                    and any(
-                        "already exists" in error
-                        for error in create_group_made_member_error.message_dict[
-                            "hashed_group_member_id"
-                        ]
-                    )
-                )
-                if not error_is_already_exists:
-                    raise
+            if not error_is_already_exists:
+                raise create_group_made_member_error from create_group_made_member_error
 
             await ctx.followup.send(content="Successfully made you a member!", ephemeral=True)
 
@@ -303,58 +240,6 @@ async def member_count(self, ctx: "TeXBotApplicationContext") -> None:  # type:
         await ctx.defer(ephemeral=False)
 
         async with ctx.typing():
-            http_session: aiohttp.ClientSession = aiohttp.ClientSession(
-                headers=REQUEST_HEADERS,
-                cookies=REQUEST_COOKIES,
-            )
-            async with http_session, http_session.get(BASE_MEMBERS_URL) 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=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"{GROUP_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=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"{GROUP_NAME} has {
-                    len(member_table.find_all('tr', {'class': ['msl_row', 'msl_altrow']}))
-                } members! :tada:"
+                content=f"{GROUP_NAME} has {await get_membership_count()} members! :tada:",
             )

pyproject.toml

@@ -25,6 +25,7 @@ test = ["pytest>=8.3"]
 type-check = ["django-stubs[compatible-mypy]>=5.1", "mypy>=1.13", "types-beautifulsoup4>=4.12"]
 
 [project] # TODO: Remove [project] table once https://github.com/astral-sh/uv/issues/8582 is completed
+dependencies = ["anyio>=4.9.0"]
 name = "TeX-Bot-Py-V2"
 requires-python = ">=3.12,<3.13" # TODO: Allow Python 3.13 once py-cord makes a new release with support for it
 version = "0.1.0"

utils/msl/__init__.py

@@ -0,0 +1,33 @@
+"""MSL utility classes & functions provided for use across the whole of the project."""
+
+from typing import TYPE_CHECKING
+
+from .events import create_event, get_all_guild_events
+from .finances import (
+    fetch_financial_transactions,
+    fetch_transaction_from_id,
+    get_account_balance,
+)
+from .memberships import get_full_membership_list, get_membership_count, is_student_id_member
+from .reports import (
+    get_product_customisations,
+    get_product_sales,
+    update_current_year_sales_report,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+__all__: "Sequence[str]" = (
+    "create_event",
+    "fetch_financial_transactions",
+    "fetch_transaction_from_id",
+    "get_account_balance",
+    "get_all_guild_events",
+    "get_full_membership_list",
+    "get_membership_count",
+    "get_product_customisations",
+    "get_product_sales",
+    "is_student_id_member",
+    "update_current_year_sales_report",
+)

utils/msl/activities.py

@@ -0,0 +1,119 @@
+"""Module for fetching activities from the guild website."""
+
+import logging
+from enum import Enum
+from typing import TYPE_CHECKING, Final
+
+import aiohttp
+import bs4
+from bs4 import BeautifulSoup
+
+from .core import BASE_HEADERS, ORGANISATION_ID, get_msl_context
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+    from datetime import datetime
+    from logging import Logger
+
+__all__: "Sequence[str]" = ()
+
+logger: "Final[Logger]" = logging.getLogger("TeX-Bot")
+
+
+ACTIVITIES_URL: Final[str] = (
+    f"https://www.guildofstudents.com/organisation/admin/activities/all/{ORGANISATION_ID}/"
+)
+
+ACTIVITIES_BUTTON_KEY: Final[str] = "ctl00$ctl00$Main$AdminPageContent$fsFilter$btnSubmit"
+ACTIVITIES_TABLE_ID: Final[str] = "ctl00_ctl00_Main_AdminPageContent_gvResults"
+ACTIVITIES_START_DATE_KEY: Final[str] = "ctl00$ctl00$Main$AdminPageContent$drDates$txtFromDate"
+ACTIVITIES_END_DATE_KEY: Final[str] = "ctl00$ctl00$Main$AdminPageContent$drDates$txtToDate"
+
+
+class ActivityStatus(Enum):
+    """
+    Enum to define the possible activity status values.
+
+    Submitted - The activity has been submitted and is pending approval.
+    Approved - The activity has been approved and is scheduled.
+    Draft - The activity is a draft and is not yet submitted.
+    Cancelled - The activity has been cancelled.
+    Queried - The activity has been queried and is pending response.
+    """
+
+    SUBMITTED = "Submitted"
+    APPROVED = "Approved"
+    DRAFT = "Draft"
+    CANCELLED = "Cancelled"
+    QUERIED = "Queried"
+
+
+async def fetch_guild_activities(from_date: "datetime", to_date: "datetime") -> dict[str, str]:
+    """Fetch all activities on the guild website."""
+    data_fields, cookies = await get_msl_context(url=ACTIVITIES_URL)
+
+    form_data: dict[str, str] = {
+        ACTIVITIES_START_DATE_KEY: from_date.strftime("%d/%m/%Y"),
+        ACTIVITIES_END_DATE_KEY: to_date.strftime("%d/%m/%Y"),
+        ACTIVITIES_BUTTON_KEY: "Apply",
+        "__EVENTTARGET": "",
+        "__EVENTARGUMENT": "",
+        "__VIEWSTATEENCRYPTED": "",
+    }
+
+    data_fields.update(form_data)
+
+    data_fields.pop("ctl00$ctl00$Main$AdminPageContent$fsFilter$btnCancel")
+
+    session_v2: aiohttp.ClientSession = aiohttp.ClientSession(
+        headers=BASE_HEADERS,
+        cookies=cookies,
+    )
+    async with (
+        session_v2,
+        session_v2.post(url=ACTIVITIES_URL, data=data_fields) as http_response,
+    ):
+        if http_response.status != 200:
+            logger.debug("Returned a non 200 status code!!")
+            logger.debug(http_response)
+            return {}
+
+        response_html: str = await http_response.text()
+
+    activities_table_html: bs4.Tag | bs4.NavigableString | None = BeautifulSoup(
+        markup=response_html,
+        features="html.parser",
+    ).find(
+        name="table",
+        attrs={"id": ACTIVITIES_TABLE_ID},
+    )
+
+    if activities_table_html is None or isinstance(activities_table_html, bs4.NavigableString):
+        logger.warning("Failed to find the activities table.")
+        logger.debug(response_html)
+        return {}
+
+    if "There are no activities" in str(activities_table_html):
+        logger.debug("No activities were found matching the date range.")
+        return {}
+
+    activities_list: list[bs4.Tag] = activities_table_html.find_all(name="tr")
+
+    activities_list.pop(0)
+
+    return {
+        activity.find(name="a").get("href").split("/")[7]: activity.find_all(name="td")[  # type: ignore[union-attr]
+            1
+        ].text.strip()
+        for activity in activities_list
+    }
+
+
+async def create_activity() -> int:
+    """Create an activity on the guild website."""
+    raise NotImplementedError
+
+
+async def fetch_activity(activity_id: int) -> dict[str, str]:
+    """Fetch a specific activity from the guild website."""
+    raise NotImplementedError