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
113
114async def _get_subject_attendance(client):
115    types = {
116        "1": "nb",
117        "2": "sp",
118        "3": "u",
119        "4": "zw",
120        "100": "ob",
121        "1266": "wy",
122        "2022": "k",
123        "2829": "sz",
124    }
125
126    client.refresh_oauth()
127    attendances = client.get(client.GATEWAY_API_ATTENDANCE).json()["Attendances"]
128
129    cookies = client._session.cookies
130    headers = client._session.headers
131    base_url = client.BASE_URL
132
133    lesson_cache = {}
134    subject_cache = {}
135
136    async def req(session: ClientSession, url: str, retries=5, delay=0.5):
137        for attempt in range(retries):
138            try:
139                async with session.get(url) as response:
140                    response.raise_for_status()
141                    return await response.json()
142            except:
143                if attempt == retries - 1:
144                    raise
145                await asyncio.sleep(delay * (2**attempt))
146
147    async def _lesson_attendance(session: ClientSession, attendance: dict):
148        lesson_id = attendance["Lesson"]["Id"]
149        absence_type = types.get(str(attendance["Type"]["Id"]), "unknown")
150
151        if lesson_id in lesson_cache:
152            return (lesson_cache[lesson_id], absence_type)
153
154        lesson_resp = await req(
155            session, f"{base_url}/gateway/api/2.0/Lessons/{lesson_id}"
156        )
157        sub_id = lesson_resp["Lesson"]["Subject"]["Id"]
158
159        if sub_id in subject_cache:
160            subject_name = subject_cache[sub_id]
161        else:
162            sub_resp = await req(
163                session, f"{base_url}/gateway/api/2.0/Subjects/{sub_id}"
164            )
165            subject_name = sub_resp["Subject"]["Name"]
166            subject_cache[sub_id] = subject_name
167
168        lesson_cache[lesson_id] = subject_name
169        return (subject_name, absence_type)
170
171    async with ClientSession(cookies=cookies, headers=headers) as session:
172        results = await asyncio.gather(
173            *[_lesson_attendance(session, attendance) for attendance in attendances]
174        )
175
176    counts = defaultdict(lambda: defaultdict(int))
177    for subject, absence in results:
178        counts[subject][absence] += 1
179
180    return {subject: dict(types) for subject, types in counts.items()}
181
182
183def get_subject_frequency(client: Client, attendances=None) -> Dict[str, float]:
184    if not attendances:
185        attendances = asyncio.run(_get_subject_attendance(client))
186    frequency = {}
187    for sub in attendances:
188        attended = attendances[sub].get("ob", 0) + attendances[sub].get("sp", 0)
189        unattended = (
190            attendances[sub].get("nb", 0) + attendances[sub].get("u", 0) + attendances[sub].get("zw", 0)
191        )
192        total = attended + unattended
193        frequency[sub] = round(attended / total * 100, 2) if total > 0 else 100.0
194    return frequency
195
196
197def get_gateway_attendance(client: Client) -> List[Tuple[Tuple[str, str], str, str]]:
198    """
199    Retrieves attendance data from the gateway API.
200
201    The gateway API data is typically updated every 3 hours.
202    Accessing api.librus.pl requires a private key.
203
204    Requires:
205        oauth token to be refreshed with client.refresh_oauth()
206
207    Args:
208        client (Client): The client object used to make the request.
209
210    Returns:
211        List[Tuple[Tuple[str, str], str, str]]: A list of tuples containing attendance data.
212            Each tuple contains three elements:
213            1. Tuple containing type abbreviation and type name.
214            2. Lesson number.
215            3. Semester.
216
217    Raises:
218        ValueError: If the OAuth token is missing.
219        AuthorizationError: If there is an authorization error while accessing the API.
220    """
221    types = {
222        "1": {"short": "nb", "name": "Nieobecność"},
223        "2": {"short": "sp", "name": "Spóźnienie"},
224        "3": {"short": "u", "name": "Nieobecność uspr."},
225        "4": {"short": "zw", "name": "Zwolnienie"},
226        "100": {"short": "ob", "name": "Obecność"},
227        "1266": {"short": "wy", "name": "Wycieczka"},
228        "2022": {"short": "k", "name": "Konkurs szkolny"},
229        "2829": {"short": "sz", "name": "Szkolenie"},
230    }
231    oauth = client.token.oauth
232    if oauth == "":
233        oauth = client.refresh_oauth()
234    client.cookies["oauth_token"] = oauth
235    response = client.get(client.GATEWAY_API_ATTENDANCE)
236
237    attendances = response.json()["Attendances"]
238    _attendance = []
239    for a in attendances:
240        type_id = a["Type"]["Id"]
241        type_data = tuple(types[str(type_id)].values())
242        lesson_number = a["LessonNo"]
243        semester = a["Semester"]
244
245        _attendance.append((type_data, lesson_number, semester))
246
247    return _attendance
248
249
250def get_attendance_frequency(client: Client) -> Tuple[float, float, float]:
251    """
252    Calculates the attendance frequency for each semester and overall.
253
254    Args:
255        client (Client): The client object used to retrieve attendance data.
256
257    Returns:
258        Tuple[float, float, float]: A tuple containing the attendance frequencies for the first semester, second semester, and overall.
259            Each frequency is a float value between 0 and 1, representing the ratio of attended lessons to total lessons.
260
261    Raises:
262        ValueError: If there is an error retrieving attendance data.
263    """
264    attendance = get_gateway_attendance(client)
265    first_semester = [a for a in attendance if a[2] == 1]
266    second_semester = [a for a in attendance if a[2] == 2]
267    f_attended = len([a for a in first_semester if a[0][0] in ["wy", "ob", "sp"]])
268    s_attended = len([a for a in second_semester if a[0][0] in ["wy", "ob", "sp"]])
269    f_freq = f_attended / len(first_semester) if len(first_semester) != 0 else 1
270    s_freq = s_attended / len(second_semester) if len(second_semester) != 0 else 1
271    overall_freq = (
272        len([a for a in attendance if a[0][0] in ["wy", "ob", "sp"]]) / len(attendance)
273        if len(attendance) != 0
274        else 1
275    )
276    return f_freq, s_freq, overall_freq
277    # ADD Lesson frequency
278
279
280def _extract_title_pairs(title: str):
281    sanitize_title = (
282        title.replace("</b>", "<br>").replace("<br/>", "").strip().split("<br>")
283    )
284
285    return [pair.split(":", 1) for pair in sanitize_title]
286
287
288def _sanitize_pairs(pairs: List[List[str]]) -> Dict[str, str]:
289    sanitized_pairs = {}
290    for pair in pairs:
291        if len(pair) != 2:
292            sanitized_pairs[pair[0].strip()] = "unknown"
293            continue
294        key, val = pair
295        sanitized_pairs[key.strip()] = val.strip()
296    return sanitized_pairs
297
298
299def _sanitize_onclick_href(onclick: str):
300    href = (
301        onclick.replace("otworz_w_nowym_oknie(", "")
302        .split(",")[0]
303        .replace("'", "")
304        .split("/")
305    )
306    if len(href) < 4:
307        return ""
308    return href[3]
309
310
311def _create_attendance(single: Tag, semester: int):
312    """
313    Creates an Attendance object from a single attendance record.
314
315    Args:
316        single (Tag): The BeautifulSoup Tag representing a single attendance record.
317        semester (int): The semester number to which the attendance record belongs.
318
319    Returns:
320        Attendance: An Attendance object representing the parsed attendance record.
321
322    Raises:
323        ParseError: If there is an error parsing the attendance record.
324    """
325    if single.attrs.get("title") is None:
326        raise ParseError("Absence anchor title is None")
327    pairs = _extract_title_pairs(single.attrs["title"])
328    attributes = _sanitize_pairs(pairs)
329
330    date = attributes.get("Data", "").split(" ")[0]
331    _type = attributes.get("Rodzaj", "")
332    school_subject = attributes.get("Lekcja", "")
333    topic = attributes.get("Temat zajęć", "")
334    period = int(attributes.get("Godzina lekcyjna", "0"))
335    excursion = True if attributes.get("Czy wycieczka", "") == "Tak" else False
336    teacher = attributes.get("Nauczyciel", "")
337
338    href = _sanitize_onclick_href(single.attrs.get("onclick", ""))
339
340    return Attendance(
341        single.text,
342        href,
343        semester,
344        date,
345        _type,
346        teacher,
347        period,
348        excursion,
349        topic,
350        school_subject,
351    )
352
353
354def get_attendance(client: Client, sort_by: str = "all") -> List[List[Attendance]]:
355    """
356    Retrieves attendance records from librus.
357
358    Args:
359        client (Client): The client object used to fetch attendance data.
360        sort_by (str, optional): The sorting criteria for attendance records.
361            It can be one of the following values:
362            - "all": Sort by all attendance records.
363            - "week": Sort by attendance records for the current week.
364            - "last_login": Sort by attendance records since the last login.
365            Defaults to "all".
366
367    Returns:
368        List[List[Attendance]]: A list containing attendance records grouped by semester.
369            Each inner list represents attendance records for a specific semester.
370
371    Raises:
372        ArgumentError: If an invalid value is provided for the sort_by parameter.
373        ParseError: If there is an error parsing the attendance data.
374    """
375    SORT: Dict[str, Dict[str, str]] = {
376        "all": {"zmiany_logowanie_wszystkie": ""},
377        "week": {"zmiany_logowanie_tydzien": "zmiany_logowanie_tydzien"},
378        "last_login": {"zmiany_logowanie": "zmiany_logowanie"},
379    }
380    if sort_by not in SORT.keys():
381        raise ArgumentError(
382            "Wrong value for sort_by it can be either all, week or last_login"
383        )
384
385    soup = no_access_check(
386        BeautifulSoup(
387            client.post(client.ATTENDANCE_URL, data=SORT[sort_by]).text,
388            "lxml",
389        )
390    )
391    table = soup.find("table", attrs={"class": "center big decorated"})
392    if table is None or isinstance(table, NavigableString):
393        raise ParseError("Error parsing attendance (table).")
394
395    days = table.find_all("tr", attrs={"class": ["line0", "line1"]})
396    attendance_semesters = [[] for _ in range(2)]  # Two semesters
397    semester = -1
398    for day in days:
399        if day.find("td", attrs={"class": "center bolded"}):
400            # marker to increment semester
401            semester += 1
402        attendance = day.find_all("td", attrs={"class": "center"})
403        for absence in attendance:
404            a_elem: List[Tag] = absence.find_all("a")
405            for single in a_elem:
406                attendance_semesters[semester].append(
407                    _create_attendance(single, semester)
408                )
409    match semester:
410        case 0:
411            return list(attendance_semesters)
412        case 1:
413            return list(reversed(attendance_semesters))
414        case _:
415            raise ParseError("Couldn't find attendance semester")