librus_apix.notifications

  1"""
  2This module provides functionality to interact with the Librus site and extract notifications from the user's dashboard.
  3
  4The notification bubbles are bound to token, and their amount doesn't change unless you retrieve a new Token, hence we have to
  5request every individual last_login endpoint and retrieve stuff from there.
  6
  7Classes:
  8    - NotificationAmount: Represents a notification with a destination and an amount.
  9    - NotificationData: Represents data of various notifications including grades, attendance, messages, announcements, schedule, and homework.
 10    - NotificationIds: Represents the IDs of various notifications to track seen notifications.
 11
 12Functions:
 13    - get_initial_notification_data(client: Client) -> Tuple[NotificationData, NotificationIds]: Fetches and parses the initial notification data and their IDs for a new token.
 14    - get_new_notification_data(client: Client, seen_notifications: NotificationIds) -> Tuple[NotificationData, NotificationIds]: Fetches and parses new notifications using NotificationIds, returns data and updates seen notification IDs.
 15"""
 16
 17from dataclasses import dataclass
 18from datetime import datetime, timedelta
 19from hashlib import md5
 20from typing import Any, DefaultDict, List, Tuple
 21
 22from bs4 import BeautifulSoup, Tag
 23
 24from librus_apix.announcements import Announcement, get_announcements
 25from librus_apix.attendance import Attendance, get_attendance
 26from librus_apix.client import Client
 27from librus_apix.exceptions import ParseError
 28from librus_apix.grades import Grade, get_grades
 29from librus_apix.helpers import no_access_check
 30from librus_apix.homework import Homework, get_homework
 31from librus_apix.messages import Message, get_received
 32from librus_apix.schedule import RecentEvent, get_recently_added_schedule
 33
 34
 35@dataclass
 36class NotificationAmount:
 37    """
 38    Represents a notification with a destination identifier name and an amount.
 39
 40    Attributes:
 41        name (str): Str representation of name ex: Oceny
 42        destination (str): endpoint
 43        amount (int): The count of notifications for the given destination.
 44    """
 45
 46    name: str
 47    destination: str
 48    amount: int
 49
 50
 51def get_new_token_notification_amounts(client: Client) -> List[NotificationAmount]:
 52    """
 53    Fetches and parses notification amounts from the user's dashboard on the Librus platform.
 54
 55    Args:
 56        client (Client): An instance of `librus_apix.client.Client` used to make requests to the Librus platform.
 57
 58    Returns:
 59        List[NotificationAmount]: A list of `NotificationAmount` objects representing the notifications found on the user's dashboard.
 60    """
 61    soup = no_access_check(BeautifulSoup(client.get(client.INDEX_URL).text, "lxml"))
 62    notifications = []
 63    circles = soup.select("div#graphic-menu > ul > li > a[class!='button counter']")
 64    for circle in circles:
 65        name = circle.text.replace("\n", "").strip()
 66        destination = circle.attrs.get("href", "/")
 67        amount = 0
 68        if (
 69            name == "Widok alternatywny"
 70            or "javascript" in destination
 71            or destination
 72            not in [
 73                "/ogloszenia",
 74                "/moje_zadania",
 75                "/wiadomosci",
 76                "/przegladaj_oceny/uczen",
 77                "/przegladaj_nb/uczen",
 78                "/terminarz",
 79            ]
 80        ):
 81            continue
 82
 83        if isinstance(circle.parent, Tag):
 84            counter = circle.parent.select_one("a.button.counter")
 85            if isinstance(counter, Tag):
 86                try:
 87                    amount = int(counter.text)
 88                except ValueError:
 89                    pass
 90
 91        notifications.append(NotificationAmount(name, destination, amount))
 92
 93    return notifications
 94
 95
 96def _compare_hrefs(object_href: str, href_ids: List[str]):
 97    return object_href in href_ids
 98
 99
100def _parse_recent_schedule_notification(
101    schedule: List[RecentEvent], seen_ids: List[str] = []
102):
103    new_schedule = []
104    for event in schedule:
105        data_bytes = event.data.encode("utf-8")
106        _id = md5(data_bytes).hexdigest()
107        if _compare_hrefs(_id, seen_ids):
108            continue
109        seen_ids.append(_id)
110        new_schedule.append(event)
111    return new_schedule, seen_ids
112
113
114def _parse_announcements_notification(
115    announcements: List[Announcement], seen_ids: List[str] = []
116):
117    new_announcements = []
118    for announcement in announcements:
119        _id = announcement.title + announcement.date
120        if _compare_hrefs(_id, seen_ids):
121            break
122        seen_ids.append(_id)
123        new_announcements.append(announcement)
124    return new_announcements, seen_ids
125
126
127def _parse_homework_notification(homework: List[Homework], seen_ids: List[str] = []):
128    new_homework = []
129    for hw in homework:
130        href = hw.href
131        if _compare_hrefs(href, seen_ids):
132            break
133        seen_ids.append(href)
134        new_homework.append(hw)
135    return new_homework, seen_ids
136
137
138def _parse_messages_notification(messages: List[Message], seen_ids: List[str] = []):
139    new_messages = []
140    new_ids = []
141    for message in messages:
142        href = message.href
143        if _compare_hrefs(href, seen_ids):
144            break
145        if message.unread == False:
146            continue
147        new_ids.append(href)
148        new_messages.append(message)
149    if len(messages) > 0 and len(seen_ids) == 0:
150        seen_ids.append(messages[0].href)
151    else:
152        seen_ids.extend(new_ids)
153    return new_messages, seen_ids
154
155
156def _parse_attendance_notification(
157    attendance: List[List[Attendance]], seen_ids: List[str] = []
158):
159    new_attendance = []
160    for semester in attendance:
161        for semester_attendance in semester:
162            href = semester_attendance.href
163            if _compare_hrefs(href, seen_ids):
164                continue
165            seen_ids.append(href)
166            new_attendance.append(semester_attendance)
167    return new_attendance, seen_ids
168
169
170def _parse_grades_notifications(
171    grades: List[DefaultDict[str, List[Grade]]], seen_ids: List[str] = []
172):
173    new_grades = []
174    for semester in grades:
175        for subject_grades in semester.values():
176            for grade in subject_grades:
177                href = grade.href
178                if _compare_hrefs(href, seen_ids):
179                    continue
180                new_grades.append(grade)
181                seen_ids.append(href)
182    return new_grades, seen_ids
183
184
185def parse_basic_amount(
186    client: Client, amount: NotificationAmount
187) -> Tuple[List[Any], List[str]]:
188    if amount.amount == 0 and amount.destination not in [
189        "/ogloszenia",
190        "/moje_zadania",
191        "/wiadomosci",
192    ]:
193        return [], []
194    match amount.destination:
195        case "/przegladaj_oceny/uczen":
196            grades, _averages, _descriptive = get_grades(client, "last_login")
197            return _parse_grades_notifications(grades)
198        case "/przegladaj_nb/uczen":
199            attendance = get_attendance(client, "last_login")
200            return _parse_attendance_notification(attendance)
201        case "/wiadomosci":
202            messages = get_received(client, 0)
203            top_two_msgs = messages[:2]
204            if len(top_two_msgs) == 0:
205                return [], []
206            else:
207                return _parse_messages_notification(top_two_msgs)
208
209        case "/ogloszenia":
210            announcements = get_announcements(client)
211            newest = announcements[: amount.amount]
212            if len(newest) == 0:
213                newest = [announcements[0]]
214                _, ids = _parse_announcements_notification(newest)
215                return [], ids
216            else:
217                return _parse_announcements_notification(newest)
218
219        case "/terminarz":
220            schedule = get_recently_added_schedule(client)
221            return schedule, []
222        case "/moje_zadania":
223            today = datetime.now()
224            hw_amount = -amount.amount
225            if hw_amount == 0:
226                hw_amount = -1
227            homework = get_homework(
228                client,
229                (today - timedelta(days=7)).strftime("%Y-%m-%d"),
230                today.strftime("%Y-%m-%d"),
231            )[hw_amount:]
232            if amount.amount == 0:
233                _, ids = _parse_homework_notification(homework)
234                return [], ids
235            else:
236                return _parse_homework_notification(homework)
237
238        case _:
239            return [], []
240
241
242@dataclass
243class NotificationData:
244    """
245    Represents data of various notifications.
246
247    Attributes:
248        grades (List[Grade]): A list of grade notifications.
249        attendance (List[Attendance]): A list of attendance notifications.
250        messages (List[Message]): A list of message notifications.
251        announcements (List[Announcement]): A list of announcement notifications.
252        schedule (List[RecentEvent]): A list of schedule notifications.
253        homework (List[Homework]): A list of homework notifications.
254    """
255
256    grades: List[Grade]
257    attendance: List[Attendance]
258    messages: List[Message]
259    announcements: List[Announcement]
260    schedule: List[RecentEvent]
261    homework: List[Homework]
262
263
264@dataclass
265class NotificationIds:
266    """
267    Represents the IDs (mostly .href) of various notifications to track seen notifications.
268
269    Attributes:
270        grades (List[str]): A list of grade notification IDs.
271        attendance (List[str]): A list of attendance notification IDs.
272        messages (List[str]): A list of message notification IDs.
273        announcements (List[str]): A list of announcement notification IDs (title+data) concat.
274        schedule (List[str]): A list of schedule notification IDs.
275        homework (List[str]): A list of homework notification IDs.
276    """
277
278    grades: List[str]
279    attendance: List[str]
280    messages: List[str]
281    announcements: List[str]
282    schedule: List[str]
283    homework: List[str]
284
285
286def get_initial_notification_data(client: Client):
287    """
288    Fetches and parses the initial notification data and their IDs for a new token.
289    ! Should only be ran once on every new Token. The notifications are stored inside Token and won't update.
290
291    Args:
292        client (Client): An instance of `librus_apix.client.Client`.
293
294    Returns:
295        Tuple[NotificationData, NotificationIds]: A tuple containing the initial notification data and their IDs.
296    """
297    amounts = get_new_token_notification_amounts(client)
298    amounts = map(lambda amount: parse_basic_amount(client, amount), amounts)
299    notify_data = []
300    notify_ids = []
301    for data, ids in amounts:
302        notify_data.append(data)
303        notify_ids.append(ids)
304
305    if len(notify_data) != 6:
306        raise ParseError("notification length doenst match expected 6")
307    return NotificationData(*notify_data), NotificationIds(*notify_ids)
308
309
310def get_new_notification_data(client: Client, seen_notifications: NotificationIds):
311    """
312    Fetches and parses new notification data and updates seen notification IDs based on given NotificationIds.
313
314    Args:
315        client (Client): An instance of `librus_apix.client.Client`.
316        seen_notifications (NotificationIds): A `NotificationIds` object representing the seen notifications.
317
318    Returns:
319        Tuple[NotificationData, NotificationIds]: A tuple containing the new notification data and updated seen notification IDs.
320    """
321    grades, _, _ = get_grades(client, "last_login")
322    attendance = get_attendance(client, "last_login")
323    messages = get_received(client, 0)
324    announcements = get_announcements(client)
325    today = datetime.now()
326    homework = get_homework(
327        client,
328        (today - timedelta(days=7)).strftime("%Y-%m-%d"),
329        today.strftime("%Y-%m-%d"),
330    )[::-1]
331
332    schedule = get_recently_added_schedule(client)
333
334    new_schedule, seen_events = _parse_recent_schedule_notification(
335        schedule, seen_notifications.schedule
336    )
337    new_grades, seen_grades = _parse_grades_notifications(
338        grades, seen_notifications.grades
339    )
340    new_attendance, seen_attendance = _parse_attendance_notification(
341        attendance, seen_notifications.attendance
342    )
343    new_messages, seen_messages = _parse_messages_notification(
344        messages, seen_notifications.messages
345    )
346    new_announcements, seen_announcements = _parse_announcements_notification(
347        announcements, seen_notifications.announcements
348    )
349    new_homework, seen_homework = _parse_homework_notification(
350        homework, seen_notifications.homework
351    )
352
353    return NotificationData(
354        new_grades,
355        new_attendance,
356        new_messages,
357        new_announcements,
358        new_schedule,
359        new_homework,
360    ), NotificationIds(
361        seen_grades,
362        seen_attendance,
363        seen_messages,
364        seen_announcements,
365        seen_events,
366        seen_homework,
367    )