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