feat: add KYCAS

app/app.py

@@ -194,6 +194,7 @@ async def authenticate(payload: RequestModel, background_tasks: BackgroundTasks)
     username = payload.username
     password = payload.password
     profile = payload.profile
+    know_your_class_and_section = payload.know_your_class_and_section
     fields = payload.fields
 
     # Authenticate the user
@@ -204,6 +205,7 @@ async def authenticate(payload: RequestModel, background_tasks: BackgroundTasks)
             username=username,
             password=password,
             profile=profile,
+            know_your_class_and_section=know_your_class_and_section,
             fields=fields,
         ),
     )
@@ -214,7 +216,7 @@ async def authenticate(payload: RequestModel, background_tasks: BackgroundTasks)
     try:
         authentication_result = ResponseModel.model_validate(authentication_result)
         logging.info(f"Returning auth result for user={username}: {authentication_result}")
-        authentication_result = authentication_result.model_dump(exclude_none=True)
+        authentication_result = authentication_result.model_dump(exclude_none=True, by_alias=True)
         authentication_result["timestamp"] = current_time.isoformat()
         return JSONResponse(
             status_code=200,

app/docs/authenticate.py

@@ -23,6 +23,16 @@
                                 "profile": True,
                             },
                         },
+                        "auth_with_class_and_section": {
+                            "summary": "Authentication with KYCAS",
+                            "description": "Authentication with Know Your Class and Section data",
+                            "value": {
+                                "username": "PES1201800001",
+                                "password": "mySecurePassword123",
+                                "profile": True,
+                                "know_your_class_and_section": True,
+                            },
+                        },
                         "phone_auth_selective_fields": {
                             "summary": "Authentication with Selected Fields",
                             "description": "Authentication using username and requesting specific profile data fields",
@@ -74,6 +84,38 @@
                                 },
                             },
                         },
+                        "authentication_with_kycas": {
+                            "summary": "Authentication with KYCAS",
+                            "value": {
+                                "status": True,
+                                "message": "Login successful.",
+                                "timestamp": "2024-07-28T22:30:10.103368+05:30",
+                                "profile": {
+                                    "name": "John Doe",
+                                    "prn": "PESXXYYZZZZZ",
+                                    "srn": "PESXXUGYYZZZ",
+                                    "program": "Bachelor of Technology",
+                                    "branch": "Computer Science and Engineering",
+                                    "semester": "2",
+                                    "section": "C",
+                                    "email": "johndoe@gmail.com",
+                                    "phone": "1234567890",
+                                    "campus_code": 1,
+                                    "campus": "RR",
+                                },
+                                "know_your_class_and_section": {
+                                    "prn": "PESXXYYZZZZZ",
+                                    "srn": "PESXXUGYYZZZ",
+                                    "name": "John Doe",
+                                    "class": "Sem-X",
+                                    "section": "Section X",
+                                    "cycle": "NA",
+                                    "department": "Computer Science and Engineering",
+                                    "branch": "CSE",
+                                    "institute_name": "PES University",
+                                },
+                            },
+                        },
                         "authentication_with_selected_fields": {
                             "summary": "Authentication with Selected Fields",
                             "value": {
@@ -167,6 +209,14 @@
                                 "timestamp": "2024-07-28T22:30:10.103368+05:30",
                             },
                         },
+                        "kycas_fetch_error": {
+                            "summary": "KYCAS page fetching failed",
+                            "value": {
+                                "status": False,
+                                "message": "Failed to fetch Know Your Class and Section data from PESU Academy.",
+                                "timestamp": "2024-07-28T22:30:10.103368+05:30",
+                            },
+                        },
                     }
                 }
             },

app/exceptions/authentication.py

@@ -33,3 +33,11 @@ class ProfileParseError(PESUAcademyError):
     def __init__(self, message: str = "Failed to parse student profile page from PESU Academy.") -> None:
         """Initialize the ProfileParseError with a custom message."""
         super().__init__(message, status_code=422)
+
+
+class KYCASFetchError(PESUAcademyError):
+    """Raised when Know Your Class and Section data could not be fetched from PESU Academy."""
+
+    def __init__(self, message: str = "Failed to fetch Know Your Class and Section data from PESU Academy.") -> None:
+        """Initialize the KYCASFetchError with a custom message."""
+        super().__init__(message, status_code=502)

app/models/__init__.py

@@ -1,5 +1,6 @@
 """Custom models for the PESUAuth API."""
 
+from .kycas import KYCASModel as KYCASModel
 from .profile import ProfileModel as ProfileModel
 from .request import RequestModel as RequestModel
 from .response import ResponseModel as ResponseModel

app/models/kycas.py

@@ -0,0 +1,66 @@
+"""Model representing the Know Your Class and Section data returned after successful authentication."""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class KYCASModel(BaseModel):
+    """Model representing the Know Your Class and Section data."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    prn: str | None = Field(
+        None,
+        title="PRN",
+        description="PRN of the user.",
+        json_schema_extra={"example": "PESXXYYZZZZZ"},
+    )
+    srn: str | None = Field(
+        None,
+        title="SRN",
+        description="SRN of the user.",
+        json_schema_extra={"example": "PESXXUGYYZZZ"},
+    )
+    name: str | None = Field(
+        None,
+        title="Name",
+        description="Full name of the user.",
+        json_schema_extra={"example": "John Doe"},
+    )
+    class_field: str | None = Field(
+        None,
+        validation_alias="class",
+        serialization_alias="class",
+        title="Class",
+        description="Class of the user.",
+        json_schema_extra={"example": "Sem-X"},
+    )
+    section: str | None = Field(
+        None,
+        title="Section",
+        description="Section the user belongs to.",
+        json_schema_extra={"example": "Section X"},
+    )
+    cycle: str | None = Field(
+        None,
+        title="Cycle",
+        description="Cycle of the user.",
+        json_schema_extra={"example": "NA"},
+    )
+    department: str | None = Field(
+        None,
+        title="Department",
+        description="Department of the user.",
+        json_schema_extra={"example": "Computer Science and Engineering"},
+    )
+    branch: str | None = Field(
+        None,
+        title="Branch",
+        description="Branch short code of the user.",
+        json_schema_extra={"example": "CSE"},
+    )
+    institute_name: str | None = Field(
+        None,
+        title="Institute Name",
+        description="Institute name of the user.",
+        json_schema_extra={"example": "PES University"},
+    )

app/models/request.py

@@ -33,6 +33,13 @@ class RequestModel(BaseModel):
         json_schema_extra={"example": True},
     )
 
+    know_your_class_and_section: bool = Field(
+        False,
+        title="Know Your Class and Section Flag",
+        description="Whether to fetch the user's class and section information.",
+        json_schema_extra={"example": True},
+    )
+
     fields: list[Literal[*PESUAcademy.DEFAULT_FIELDS]] | None = Field(
         None,
         title="Profile Fields",

app/models/response.py

@@ -4,7 +4,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from app.models import ProfileModel
+from app.models import KYCASModel, ProfileModel
 
 
 class ResponseModel(BaseModel):
@@ -38,3 +38,10 @@ class ResponseModel(BaseModel):
         title="User Profile Data",
         description="The user's profile data returned only if authentication succeeds and profile data was requested.",
     )
+
+    know_your_class_and_section: KYCASModel | None = Field(
+        None,
+        title="Know Your Class and Section Data",
+        description="The user's class and section data returned only if authentication succeeds"
+        " and class/section data was requested.",
+    )

app/pesu.py

@@ -12,6 +12,7 @@
 from app.exceptions.authentication import (
     AuthenticationError,
     CSRFTokenError,
+    KYCASFetchError,
     ProfileFetchError,
     ProfileParseError,
 )
@@ -46,6 +47,10 @@ class PESUAcademy:
         "phone",
         "campus_code",
         "campus",
+        "class",
+        "cycle",
+        "department",
+        "institute_name",
     ]
 
     PROFILE_PAGE_HEADER_TO_KEY_MAP = {
@@ -58,6 +63,18 @@ class PESUAcademy:
         "Section": "section",
     }
 
+    KYCAS_HEADER_TO_KEY_MAP = {
+        "PRN": "prn",
+        "SRN": "srn",
+        "Name": "name",
+        "Class": "class",
+        "Section": "section",
+        "Cycle": "cycle",
+        "Department": "department",
+        "Branch": "branch",
+        "Institute Name": "institute_name",
+    }
+
     def __init__(self) -> None:
         """Initialize the PESUAcademy class."""
         self._csrf_token: str | None = None
@@ -254,11 +271,92 @@ async def get_profile_information(
 
         return profile
 
+    async def get_know_your_class_and_section(
+        self,
+        client: httpx.AsyncClient,
+        csrf_token: str,
+        username: str,
+    ) -> dict[str, Any]:
+        """Get the class and section information of the user from the Know Your Class and Section page.
+
+        Args:
+            client (httpx.AsyncClient): The authenticated HTTP client to use for making requests.
+            csrf_token (str): The authenticated CSRF token.
+            username (str): The username of the user, usually their SRN or PRN.
+
+        Returns:
+            dict[str, Any]: A dictionary containing the user's class and section information.
+        """
+        logging.info(f"Fetching class and section data for user={username} from KYCAS page...")
+        kycas_url = "https://www.pesuacademy.com/Academy/a/getStudentClassInfo"
+        kycas_data = {"controllerMode": "370", "actionType": "174", "loginId": username}
+        kycas_headers = {
+            "origin": "https://www.pesuacademy.com",
+            "referer": "https://www.pesuacademy.com/Academy/",
+            "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
+            "x-csrf-token": csrf_token,
+            "x-requested-with": "XMLHttpRequest",
+        }
+
+        try:
+            response = await client.post(kycas_url, data=kycas_data, headers=kycas_headers)
+        except Exception:
+            raise KYCASFetchError(
+                f"Failed to send KYCAS request to PESU Academy for user={username}.",
+            )
+
+        if response.status_code != 200:
+            raise KYCASFetchError(
+                f"Failed to fetch KYCAS data from PESU Academy for user={username}. "
+                f"Received status code {response.status_code}.",
+            )
+
+        soup = await asyncio.to_thread(HTMLParser, response.text)
+        kycas: dict[str, Any] = {}
+
+        table = soup.css_first("table")
+        if not table:
+            raise KYCASFetchError(
+                f"Could not find KYCAS table in the response for user={username}.",
+            )
+
+        headers = [th.text(strip=True) for th in table.css("thead th")]
+        if not headers:
+            raise KYCASFetchError(
+                f"Could not find KYCAS table headers in the response for user={username}.",
+            )
+
+        row = table.css_first("tbody tr")
+        if not row:
+            raise KYCASFetchError(
+                f"Could not find KYCAS data row in the response for user={username}.",
+            )
+
+        cells = [td.text(strip=True) for td in row.css("td")]
+
+        if len(headers) != len(cells):
+            raise KYCASFetchError(
+                f"Mismatch between KYCAS table headers ({len(headers)}) and cells ({len(cells)}) for user={username}.",
+            )
+
+        for header, cell_value in zip(headers, cells):
+            if mapped_key := self.KYCAS_HEADER_TO_KEY_MAP.get(header):
+                kycas[mapped_key] = cell_value
+
+        if not kycas:
+            raise KYCASFetchError(
+                f"No KYCAS data could be extracted for user={username}.",
+            )
+
+        logging.info(f"KYCAS data retrieved for user={username}: {kycas}.")
+        return kycas
+
     async def authenticate(
         self,
         username: str,
         password: str,
         profile: bool = False,
+        know_your_class_and_section: bool = False,
         fields: list[str] | None = None,
     ) -> dict[str, Any]:
         """Authenticate the user with the provided username and password.
@@ -267,6 +365,8 @@ async def authenticate(
             username (str): The username of the user, usually their PRN/email/phone number.
             password (str): The password of the user.
             profile (bool, optional): Whether to fetch the profile information or not. Defaults to False.
+            know_your_class_and_section (bool, optional): Whether to fetch the class and section
+                information or not. Defaults to False.
             fields (Optional[list[str]], optional): The fields to fetch from the profile.
             Defaults to None, which means all default fields will be fetched.
 
@@ -333,6 +433,24 @@ async def authenticate(
                     f"Field filtering enabled. Filtered profile data for user={username}: {result['profile']}",
                 )
 
+        if know_your_class_and_section:
+            logging.info(f"KYCAS data requested for user={username}. Fetching KYCAS data...")
+            # Fetch the class and section information
+            result["know_your_class_and_section"] = await self.get_know_your_class_and_section(
+                client,
+                csrf_token,
+                username,
+            )
+            # Filter the fields if field filtering is enabled
+            if field_filtering:
+                result["know_your_class_and_section"] = {
+                    key: value for key, value in result["know_your_class_and_section"].items() if key in fields
+                }
+                logging.info(
+                    f"Field filtering enabled. Filtered KYCAS data for user={username}: "
+                    f"{result['know_your_class_and_section']}",
+                )
+
         logging.info(f"Authentication process for user={username} completed successfully.")
 
         # Close the client and return the result