diff --git a/README.md b/README.md index bc28d0d..fe20621 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,7 @@ object, with the user's profile information if requested. | `username` | No | `str` | | The user's SRN or PRN | | `password` | No | `str` | | The user's password | | `profile` | Yes | `boolean` | `False` | Whether to fetch profile information | +| `know_your_class_and_section` | Yes | `boolean` | `False` | Whether to fetch Know Your Class and Section information | | `fields` | Yes | `list[str]` | `None` | Which fields to fetch from the profile information. If not provided, all fields will be fetched | #### Response Object @@ -123,6 +124,7 @@ profile data was requested, the response's `profile` key will store a dictionary |-------------|-----------------|--------------------------------------------------------------------------| | `status` | `boolean` | A flag indicating whether the overall request was successful | | `profile` | `ProfileObject` | A nested map storing the profile information, returned only if requested | +| `know_your_class_and_section` | `KnowYourClassAndSectionObject` | A nested map storing the profile information from PESU's Know Your Class and Section Portal | | `message` | `str` | A message that provides information corresponding to the status | | `timestamp` | `datetime` | A timezone offset timestamp indicating the time of authentication | @@ -145,6 +147,22 @@ If the authentication fails, this field will not be present in the response. | `campus_code` | The integer code of the campus (1 for RR and 2 for EC) | | `campus` | Abbreviation of the user's campus name | +#### KnowYourClassAndSectionObject + +| **Field** | **Description** | +|------------------|----------------------------------------------------------------| +| `prn` | PRN of the user | +| `srn` | SRN of the user | +| `name` | Name of the user | +| `semester` | Current semester that the user is in | +| `section` | Section of the user | +| `cycle` | Physics Cycle or Chemistry Cycle, if the user is in first year | +| `department` | Abbreviation of the branch along with the campus of the user | +| `branch` | Abbreviation of the branch that the user is pursuing | +| `institute_name` | The name of the campus that the user is studying in | +| `error` | The error name and stack trace, if an error occurs | + + ### `/health` This endpoint can be used to check the health of the API. It's useful for monitoring and uptime checks. This endpoint @@ -177,6 +195,7 @@ data = { "username": "your SRN or PRN here", "password": "your password here", "profile": True, # Optional, defaults to False + 'know_your_class_and_section': True, # Optional, defaults to False } response = requests.post("http://localhost:5000/authenticate", json=data) @@ -202,6 +221,17 @@ print(response.json()) "campus": "RR" }, "message": "Login successful.", + "know_your_class_and_section": { + "prn": "PES1201800001", + "srn": "PES1201800001", + "name": "JOHNNY BLAZE", + "semester": "Sem-8", + "section": "Section F", + "cycle": "NA", + "department": "CSE(EC Campus)", + "branch": "CSE", + "institute_name": "PES University (Electronic City)" + }, "timestamp": "2024-07-28 22:30:10.103368+05:30" } ``` diff --git a/app/app.py b/app/app.py index 91e8ecc..5fc5ece 100644 --- a/app/app.py +++ b/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, ), ) diff --git a/app/docs/authenticate.py b/app/docs/authenticate.py index 33d6521..28dbdea 100644 --- a/app/docs/authenticate.py +++ b/app/docs/authenticate.py @@ -23,6 +23,16 @@ "profile": True, }, }, + "auth_with_kycas": { + "summary": 'Authentication with "Know Your Class and Section" endpoint', + "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 "Know Your Class and Section endpoint"', + "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", + "semester": "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": '"Know Your Class and Section" endpoint 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", + }, + }, } } }, diff --git a/app/exceptions/authentication.py b/app/exceptions/authentication.py index 03833c8..62560d8 100644 --- a/app/exceptions/authentication.py +++ b/app/exceptions/authentication.py @@ -33,3 +33,14 @@ 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 "Know Your Class and Section" FetchError with a custom message.""" + super().__init__(message, status_code=502) diff --git a/app/models/__init__.py b/app/models/__init__.py index c06043f..3791feb 100644 --- a/app/models/__init__.py +++ b/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 diff --git a/app/models/kycas.py b/app/models/kycas.py new file mode 100644 index 0000000..333fdb3 --- /dev/null +++ b/app/models/kycas.py @@ -0,0 +1,64 @@ +"""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(strict=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"}, + ) + semester: str | None = Field( + None, + title="Semester", + description="Semester the user belongs to.", + 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 the user belongs to.", + json_schema_extra={"example": "NA"}, + ) + department: str | None = Field( + None, + title="Department", + description="Department the user belongs to.", + 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 the user belongs to.", + json_schema_extra={"example": "PES University"}, + ) diff --git a/app/models/profile.py b/app/models/profile.py index 7b47b96..9deedb6 100644 --- a/app/models/profile.py +++ b/app/models/profile.py @@ -43,7 +43,7 @@ class ProfileModel(BaseModel): semester: str | None = Field( None, title="Semester", - description="Current semester of the user.", + description="Current semester the user belongs to.", json_schema_extra={"example": "2"}, ) section: str | None = Field( diff --git a/app/models/request.py b/app/models/request.py index eba390c..f0b1f83 100644 --- a/app/models/request.py +++ b/app/models/request.py @@ -33,6 +33,16 @@ 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 from the " + '"Know Your Class and Section" endpoint.' + ), + json_schema_extra={"example": True}, + ) + fields: list[Literal[*PESUAcademy.DEFAULT_FIELDS]] | None = Field( None, title="Profile Fields", diff --git a/app/models/response.py b/app/models/response.py index 686e105..e7a5b6a 100644 --- a/app/models/response.py +++ b/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,12 @@ 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 from the " + '"Know Your Class and Section" endpoint returned only if authentication succeeds.' + ), + ) diff --git a/app/pesu.py b/app/pesu.py index 2f1941a..0f6ea54 100644 --- a/app/pesu.py +++ b/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", + "semester", + "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": "semester", + "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" endpoint. + + 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 from the + "Know Your Class and Section" endpoint 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