librus_apix.attendance

  1"""
  2This module provides functions for retrieving attendance records from the Librus site, parsing them, and calculating attendance frequency.
  3
  4Classes:
  5    - Attendance: Represents an attendance record with various attributes such as type, date, teacher, etc.
  6
  7Functions:
  8    - get_detail: Retrieves attendance details from a specific URL suffix.
  9    - get_gateway_attendance: Retrieves attendance data from the Librus gateway API.
 10    - get_attendance_frequency: Calculates attendance frequency for each semester and overall.
 11    - get_attendance: Retrieves attendance records from Librus based on specified sorting criteria.
 12
 13Usage:
 14```python
 15from librus_apix.client import new_client
 16
 17# Create a new client instance
 18client = new_client()
 19client.get_token(username, password)
 20
 21# Retrieve attendance details
 22detail_url = "example_detail_url"
 23attendance_details = get_detail(client, detail_url)
 24
 25# Retrieve attendance data from the gateway API
 26gateway_attendance = get_gateway_attendance(client)
 27
 28# Calculate attendance frequency
 29first_sem_freq, second_sem_freq, overall_freq = get_attendance_frequency(client)
 30
 31# Retrieve attendance records sorted by a specific criteria
 32attendance_records = get_attendance(client, sort_by="all")
 33```
 34"""
 35
 36import asyncio
 37from collections import defaultdict
 38from collections.abc import Coroutine
 39from dataclasses import dataclass
 40from typing import Any, Dict, List, Tuple
 41
 42from aiohttp import ClientSession
 43from bs4 import BeautifulSoup, NavigableString, Tag
 44
 45from librus_apix.client import Client
 46from librus_apix.exceptions import ArgumentError, ParseError
 47from librus_apix.helpers import no_access_check
 48
 49
 50@dataclass
 51class Attendance:
 52    """
 53    Represents an attendance record.
 54
 55    Attributes:
 56        symbol (str): The symbol representing the attendance record.
 57        href (str): The URL associated with the attendance record.
 58        semester (int): The semester number to which the attendance record belongs.
 59        date (str): The date of the attendance record.
 60        type (str): The type of attendance (e.g., absence, presence).
 61        teacher (str): The name of the teacher associated with the attendance record.
 62        period (int): The period or hour of the attendance record.
 63        excursion (bool): Indicates if the attendance record is related to an excursion.
 64        topic (str): The topic or subject of the attendance record.
 65        subject (str): The school subject associated with the attendance record.
 66    """
 67
 68    symbol: str
 69    href: str
 70    semester: int
 71    date: str
 72    type: str
 73    teacher: str
 74    period: int
 75    excursion: bool
 76    topic: str
 77    subject: str
 78
 79
 80def get_detail(client: Client, detail_url: str) -> Dict[str, str]:
 81    """
 82    Retrieves attendance details from the specified detail URL suffix.
 83
 84    Args:
 85        client (Client): The client object used to make the request.
 86        detail_url (str): The URL for fetching the attendance details.
 87
 88    Returns:
 89        Dict[str, str]: A dictionary containing the attendance details.
 90
 91    Raises:
 92        ParseError: If there is an error parsing the attendance details.
 93    """
 94    details = {}
 95    div = no_access_check(
 96        BeautifulSoup(
 97            client.get(client.ATTENDANCE_DETAILS_URL + detail_url).text, "lxml"
 98        )
 99    ).find("div", attrs={"class": "container-background"})
100    if div is None or isinstance(div, NavigableString):
101        raise ParseError("Error in parsing attendance details")
102    line = div.find_all("tr", attrs={"class": ["line0", "line1"]})
103    if len(line) < 1:
104        raise ParseError("Error in parsing attendance details (Lines are empty).")
105    for l in line:
106        th = l.find("th")
107        td = l.find("td")
108        if th is None or td is None:
109            continue
110        details[l.find("th").text] = l.find("td").text
111    return details
112
113def _get_subject_attendance(client: Client):
114    types = {
115        "1": {"short": "nb", "name": "Nieobecność"},
116        "2": {"short": "sp", "name": "Spóźnienie"},
117        "3": {"short": "u", "name": "Nieobecność uspr."},
118        "4": {"short": "zw", "name": "Zwolnienie"},
119        "100": {"short": "ob", "name": "Obecność"},
120        "1266": {"short": "wy", "name": "Wycieczka"},
121        "2022": {"short": "k", "name": "Konkurs szkolny"},
122        "2829": {"short": "sz", "name": "Szkolenie"},
123    }
124    client.refresh_oauth()
125    captured_lessons = {}
126    cookies = client._session.cookies
127    headers = client._session.headers
128    async def req(url, retries=5, delay=0.5) -> Coroutine[Any, Any, Any]:
129        async with ClientSession(cookies=cookies, headers=headers) as session:
130            for attempt in range(retries):
131                try:
132                    async with session.get(url) as response:
133                        response.raise_for_status()
134                        return await response.json()
135                except:
136                    if attempt == retries - 1:
137                        raise
138                    await asyncio.sleep(delay * (2 ** attempt))
139        raise
140
141    async def _lesson_attendance(attendance):
142        lesson_id = attendance["Lesson"]["Id"]
143        absence_type = types[str(attendance["Type"]["Id"])]["short"]
144        if lesson := captured_lessons.get(lesson_id, None):
145            return (lesson, absence_type)
146        resp = await req(f"{client.BASE_URL}/gateway/api/2.0/Lessons/{lesson_id}")
147        sub_id = resp["Lesson"]['Subject']['Id']
148        resp = await req(
149            f"{client.BASE_URL}/gateway/api/2.0/Subjects/{sub_id}"
150        )
151        captured_lessons[lesson_id] = resp["Subject"]["Name"]
152        return (resp["Subject"]["Name"], absence_type)
153
154
155
156    attendances = client.get(client.GATEWAY_API_ATTENDANCE).json()["Attendances"]
157    async def gather():
158        return await asyncio.gather(*[_lesson_attendance(attendance) for attendance in attendances])
159    result = asyncio.run(gather())
160    counts = defaultdict(lambda: defaultdict(int))
161    for subject, absence in result:
162        counts[subject][absence] += 1
163    result = {key: dict(value) for key, value in counts.items()}
164    return result
165
166
167def get_subject_frequency(client: Client) -> Dict[str, float]:
168    res = _get_subject_attendance(client)
169    frequency = {}
170    for sub in res:
171        attended = res[sub].get("ob", 0) + res[sub].get("sp", 0)
172        unattended = (
173            res[sub].get("nb", 0) + res[sub].get("u", 0) + res[sub].get("zw", 0)
174        )
175        total = attended + unattended
176        frequency[sub] = round(attended/total*100, 2) if total > 0 else 100.0
177    return frequency
178
179
180def get_gateway_attendance(client: Client) -> List[Tuple[Tuple[str, str], str, str]]:
181    """
182    Retrieves attendance data from the gateway API.
183
184    The gateway API data is typically updated every 3 hours.
185    Accessing api.librus.pl requires a private key.
186
187    Requires:
188        oauth token to be refreshed with client.refresh_oauth()
189
190    Args:
191        client (Client): The client object used to make the request.
192
193    Returns:
194        List[Tuple[Tuple[str, str], str, str]]: A list of tuples containing attendance data.
195            Each tuple contains three elements:
196            1. Tuple containing type abbreviation and type name.
197            2. Lesson number.
198            3. Semester.
199
200    Raises:
201        ValueError: If the OAuth token is missing.
202        AuthorizationError: If there is an authorization error while accessing the API.
203    """
204    types = {
205        "1": {"short": "nb", "name": "Nieobecność"},
206        "2": {"short": "sp", "name": "Spóźnienie"},
207        "3": {"short": "u", "name": "Nieobecność uspr."},
208        "4": {"short": "zw", "name": "Zwolnienie"},
209        "100": {"short": "ob", "name": "Obecność"},
210        "1266": {"short": "wy", "name": "Wycieczka"},
211        "2022": {"short": "k", "name": "Konkurs szkolny"},
212        "2829": {"short": "sz", "name": "Szkolenie"},
213    }
214    oauth = client.token.oauth
215    if oauth == "":
216        oauth = client.refresh_oauth()
217    client.cookies["oauth_token"] = oauth
218    response = client.get(client.GATEWAY_API_ATTENDANCE)
219
220    attendances = response.json()["Attendances"]
221    _attendance = []
222    for a in attendances:
223        type_id = a["Type"]["Id"]
224        type_data = tuple(types[str(type_id)].values())
225        lesson_number = a["LessonNo"]
226        semester = a["Semester"]
227
228        _attendance.append((type_data, lesson_number, semester))
229
230    return _attendance
231
232
233def get_attendance_frequency(client: Client) -> Tuple[float, float, float]:
234    """
235    Calculates the attendance frequency for each semester and overall.
236
237    Args:
238        client (Client): The client object used to retrieve attendance data.
239
240    Returns:
241        Tuple[float, float, float]: A tuple containing the attendance frequencies for the first semester, second semester, and overall.
242            Each frequency is a float value between 0 and 1, representing the ratio of attended lessons to total lessons.
243
244    Raises:
245        ValueError: If there is an error retrieving attendance data.
246    """
247    attendance = get_gateway_attendance(client)
248    first_semester = [a for a in attendance if a[2] == 1]
249    second_semester = [a for a in attendance if a[2] == 2]
250    f_attended = len([a for a in first_semester if a[0][0] in ["wy", "ob", "sp"]])
251    s_attended = len([a for a in second_semester if a[0][0] in ["wy", "ob", "sp"]])
252    f_freq = f_attended / len(first_semester) if len(first_semester) != 0 else 1
253    s_freq = s_attended / len(second_semester) if len(second_semester) != 0 else 1
254    overall_freq = (
255        len([a for a in attendance if a[0][0] in ["wy", "ob", "sp"]]) / len(attendance)
256        if len(attendance) != 0
257        else 1
258    )
259    return f_freq, s_freq, overall_freq
260    # ADD Lesson frequency
261
262
263def _extract_title_pairs(title: str):
264    sanitize_title = (
265        title.replace("</b>", "<br>").replace("<br/>", "").strip().split("<br>")
266    )
267
268    return [pair.split(":", 1) for pair in sanitize_title]
269
270
271def _sanitize_pairs(pairs: List[List[str]]) -> Dict[str, str]:
272    sanitized_pairs = {}
273    for pair in pairs:
274        if len(pair) != 2:
275            sanitized_pairs[pair[0].strip()] = "unknown"
276            continue
277        key, val = pair
278        sanitized_pairs[key.strip()] = val.strip()
279    return sanitized_pairs
280
281
282def _sanitize_onclick_href(onclick: str):
283    href = (
284        onclick.replace("otworz_w_nowym_oknie(", "")
285        .split(",")[0]
286        .replace("'", "")
287        .split("/")
288    )
289    if len(href) < 4:
290        return ""
291    return href[3]
292
293
294def _create_attendance(single: Tag, semester: int):
295    """
296    Creates an Attendance object from a single attendance record.
297
298    Args:
299        single (Tag): The BeautifulSoup Tag representing a single attendance record.
300        semester (int): The semester number to which the attendance record belongs.
301
302    Returns:
303        Attendance: An Attendance object representing the parsed attendance record.
304
305    Raises:
306        ParseError: If there is an error parsing the attendance record.
307    """
308    if single.attrs.get("title") is None:
309        raise ParseError("Absence anchor title is None")
310    pairs = _extract_title_pairs(single.attrs["title"])
311    attributes = _sanitize_pairs(pairs)
312
313    date = attributes.get("Data", "").split(" ")[0]
314    _type = attributes.get("Rodzaj", "")
315    school_subject = attributes.get("Lekcja", "")
316    topic = attributes.get("Temat zajęć", "")
317    period = int(attributes.get("Godzina lekcyjna", "0"))
318    excursion = True if attributes.get("Czy wycieczka", "") == "Tak" else False
319    teacher = attributes.get("Nauczyciel", "")
320
321    href = _sanitize_onclick_href(single.attrs.get("onclick", ""))
322
323    return Attendance(
324        single.text,
325        href,
326        semester,
327        date,
328        _type,
329        teacher,
330        period,
331        excursion,
332        topic,
333        school_subject,
334    )
335
336
337def get_attendance(client: Client, sort_by: str = "all") -> List[List[Attendance]]:
338    """
339    Retrieves attendance records from librus.
340
341    Args:
342        client (Client): The client object used to fetch attendance data.
343        sort_by (str, optional): The sorting criteria for attendance records.
344            It can be one of the following values:
345            - "all": Sort by all attendance records.
346            - "week": Sort by attendance records for the current week.
347            - "last_login": Sort by attendance records since the last login.
348            Defaults to "all".
349
350    Returns:
351        List[List[Attendance]]: A list containing attendance records grouped by semester.
352            Each inner list represents attendance records for a specific semester.
353
354    Raises:
355        ArgumentError: If an invalid value is provided for the sort_by parameter.
356        ParseError: If there is an error parsing the attendance data.
357    """
358    SORT: Dict[str, Dict[str, str]] = {
359        "all": {"zmiany_logowanie_wszystkie": ""},
360        "week": {"zmiany_logowanie_tydzien": "zmiany_logowanie_tydzien"},
361        "last_login": {"zmiany_logowanie": "zmiany_logowanie"},
362    }
363    if sort_by not in SORT.keys():
364        raise ArgumentError(
365            "Wrong value for sort_by it can be either all, week or last_login"
366        )
367
368    soup = no_access_check(
369        BeautifulSoup(
370            client.post(client.ATTENDANCE_URL, data=SORT[sort_by]).text,
371            "lxml",
372        )
373    )
374    table = soup.find("table", attrs={"class": "center big decorated"})
375    if table is None or isinstance(table, NavigableString):
376        raise ParseError("Error parsing attendance (table).")
377
378    days = table.find_all("tr", attrs={"class": ["line0", "line1"]})
379    attendance_semesters = [[] for _ in range(2)]  # Two semesters
380    semester = -1
381    for day in days:
382        if day.find("td", attrs={"class": "center bolded"}):
383            # marker to increment semester
384            semester += 1
385        attendance = day.find_all("td", attrs={"class": "center"})
386        for absence in attendance:
387            a_elem: List[Tag] = absence.find_all("a")
388            for single in a_elem:
389                attendance_semesters[semester].append(
390                    _create_attendance(single, semester)
391                )
392    match semester:
393        case 0:
394            return list(attendance_semesters)
395        case 1:
396            return list(reversed(attendance_semesters))
397        case _:
398            raise ParseError("Couldn't find attendance semester")