librus_apix.messages

  1"""
  2This module provides functions for interacting with messages in the Librus messaging system, including sending, retrieving, and parsing messages.
  3
  4Classes:
  5    - Message: Represents a message with details like author, title, date, etc.
  6    - MessageData: Represents the data of a message content.
  7
  8Functions:
  9    - recipient_groups: Retrieves the list of recipient groups available for sending messages.
 10    - get_recipients: Retrieves the recipients belonging to a specific group.
 11    - send_message: Sends a message to selected recipients.
 12    - message_content: Retrieves the content of a message.
 13    - get_max_page_number: Retrieves the maximum page number of messages.
 14    - get_received: Retrieves received messages from a specific page.
 15    - get_sent: Retrieves sent messages from a specific page.
 16
 17Usage:
 18    ```py
 19
 20        from librus_apix.client import new_client
 21
 22        # Create a new client instance
 23        client = new_client()
 24        client.get_token(username, password)
 25
 26        # Retrieve recipient groups and recipients
 27        groups = recipient_groups(client)
 28        recipients = get_recipients(client, groups[0])
 29
 30        # Send a message
 31        title = "Test Message"
 32        content = "This is a test message."
 33        recipient_ids = list(recipients.values())
 34        success, result_message = send_message(client, title, content, recipient_ids)
 35
 36        # Get received/sent messages
 37        messages = get_sent(client, page=1)
 38        messages = get_received(client, page=1)
 39        ...
 40        # Retrieve content of a message
 41        for message in messages:
 42            content = message_content(client, message.href)
 43            ...
 44    ```
 45"""
 46
 47from typing import List, Tuple
 48from bs4 import BeautifulSoup, Tag
 49from librus_apix.client import Client
 50from librus_apix.exceptions import ParseError
 51from librus_apix.helpers import no_access_check
 52from dataclasses import dataclass
 53import re
 54
 55
 56@dataclass
 57class MessageData:
 58    """
 59    Represents the data of a message content.
 60
 61    Attributes:
 62        author (str): The author of the message.
 63        title (str): The title of the message.
 64        content (str): The content of the message.
 65        date (str): The date when the message was sent.
 66    """
 67
 68    author: str
 69    title: str
 70    content: str
 71    date: str
 72
 73
 74@dataclass
 75class Message:
 76    """
 77    Represents a message.
 78
 79    Attributes:
 80        author (str): The author of the message.
 81        title (str): The title of the message.
 82        date (str): The date when the message was sent.
 83        href (str): The URL reference to the message.
 84        unread (bool): Indicates if the message is unread.
 85        has_attachment (bool): Indicates if the message has attachments.
 86    """
 87
 88    author: str
 89    title: str
 90    date: str
 91    href: str
 92    unread: bool
 93    has_attachment: bool
 94
 95
 96def recipient_groups(client: Client) -> List[str]:
 97    """
 98    Retrieves the list of recipient groups available for sending messages.
 99
100    Args:
101        client (Client): The client object for making HTTP requests.
102
103    Returns:
104        List[str]: A list of recipient group identifiers.
105    """
106    soup = no_access_check(
107        BeautifulSoup(client.get(client.RECIPIENT_GROUPS_URL).text, "lxml")
108    )
109    groups = []
110    trs = soup.select("table.message-recipients > tbody > tr")
111    for tr in trs:
112        radio = tr.select_one("input.recipiantTypeRadio")
113        if radio is None:
114            raise ParseError("Error getting groups (radio)")
115        groups.append(radio.attrs.get("value", ""))
116    return groups
117
118
119def get_recipients(client: Client, group: str):
120    """
121    Retrieves the recipients belonging to a specific group.
122
123    Args:
124        client (Client): The client object for making HTTP requests.
125        group (str): The identifier of the recipient group.
126
127    Returns:
128        dict: A dictionary mapping teacher names to their IDs.
129    """
130    payload = {
131        "typAdresata": group,
132        "poprzednia": "5",
133        "tabZaznaczonych": "",
134        "czyWirtualneKlasy": False,
135        "idGrupy": "0",
136    }
137    soup = no_access_check(
138        BeautifulSoup(client.post(client.RECIPIENTS_URL, data=payload).text, "lxml")
139    )
140    labels = soup.select("label")
141    teachers = {}
142    for label in labels:
143        teachers[label.text.replace("\xa0", "")] = label.attrs.get("for", "_").split(
144            "_"
145        )[-1]
146    return teachers
147
148
149def send_message(
150    client: Client,
151    title: str,
152    content: str,
153    recipient_ids: list[str],
154) -> Tuple[bool, str]:
155    """
156    Sends a message to selected recipients.
157
158    Args:
159        client (Client): The client object for making HTTP requests.
160        title (str): The title of the message.
161        content (str): The content of the message.
162        recipient_ids (list[str]): The list of recipient IDs.
163
164    Returns:
165        Tuple[bool, str]: A tuple indicating whether the message was sent successfully
166        and the result message.
167    """
168    payload = {
169        "filtrUzytkownikow": "0",
170        "idPojemnika": "",
171        "DoKogo": recipient_ids,
172        "Rodzaj": "0",
173        "temat": title,
174        "tresc": content,
175        "poprzednia": "5",
176        "fileStorageIdentifier": "",
177        "wyslij": "Wyślij",
178    }
179    sent_message = no_access_check(
180        BeautifulSoup(client.post(client.SEND_MESSAGE_URL, data=payload).text, "lxml")
181    )
182    result = sent_message.select_one("div.container-background > p")
183    if result is None:
184        raise ParseError("Error getting the result of the message!")
185    result = result.text
186    if "nie zostala" in result:
187        return False, result
188    if sent_message.status_code == 200:
189        return True, result
190    return False, result
191
192
193def unwrap_message_data(tr: Tag) -> str:
194    value = tr.select_one("td[class='left']")
195    return value.text if value is not None else ""
196
197
198def message_content(client: Client, content_url: str) -> MessageData:
199    """
200    Retrieves the content of a message.
201
202    Args:
203        client (Client): The client object for making HTTP requests.
204        content_url (str): The URL of the message content.
205
206    Returns:
207        MessageData: An object containing the message details.
208    """
209    soup = no_access_check(
210        BeautifulSoup(client.get(client.MESSAGE_URL + "/" + content_url).text, "lxml")
211    )
212    message_data = soup.select_one("table[class='stretch']")
213    if message_data is None:
214        raise ParseError("Error in parsing message data.")
215    trs = message_data.select("tr")
216    if len(trs) < 3:
217        raise ParseError("Not enough values to unpack from message_data")
218    author, title, date = trs[:3]
219    content = soup.find("div", attrs={"class": "container-message-content"})
220    if content is None:
221        raise ParseError("Error in parsing message content.")
222    return MessageData(
223        unwrap_message_data(author),
224        unwrap_message_data(title),
225        content.text,
226        unwrap_message_data(date),
227    )
228
229
230def _sanitize_href(href: str) -> str:
231    if len(href) > 4:
232        return href.split("/")[4]
233    return ""
234
235
236def parse_sent(message_soup: BeautifulSoup) -> List[Message]:
237    """
238    Parses sent messages from the message soup.
239
240    Args:
241        message_soup (BeautifulSoup): The BeautifulSoup object containing message data.
242
243    Returns:
244        List[Message]: A list of Message objects representing sent messages.
245    """
246    msgs: List[Message] = []
247    hasAttachment = False
248    soup = message_soup.find("table", attrs={"class": "decorated stretch"})
249    if soup is None:
250        raise ParseError("Error in parsing messages.")
251    tbody = soup.find("tbody")
252    if not isinstance(tbody, Tag):
253        raise ParseError("Error in parsing messages (tbody).")
254    tds = tbody.find_all("tr", attrs={"class": ["line0", "line1"]})
255    if tds[0].text.strip() == "Brak wiadomości":
256        return []
257    for td in tds:
258        hasAttachment = False
259        message_data: List[Tag] = td.find_all("td")
260        if len(message_data) < 7:
261            raise ParseError("Message data has less than 7 elements")
262        _tick, attachment, author, title, date, unread, _trash = message_data[:7]
263        if attachment.find("img"):
264            hasAttachment = True
265        unread = True if unread == "NIE" else False
266        author_a = author.find("a")
267        href = ""
268        if isinstance(author_a, Tag):
269            href = author_a.attrs.get("href", "")
270            href = _sanitize_href(href)
271        author = author.text
272        title = title.text
273        date = date.text
274        m = Message(author, title, date, href, unread, hasAttachment)
275        msgs.append(m)
276    return msgs
277
278
279def parse(message_soup: BeautifulSoup) -> List[Message]:
280    """
281    Parses received messages from the message soup.
282
283    Args:
284        message_soup (BeautifulSoup): The BeautifulSoup object containing message data.
285
286    Returns:
287        List[Message]: A list of Message objects representing received messages.
288    """
289    msgs: List[Message] = []
290    hasAttachment = False
291    soup = message_soup.find("table", attrs={"class": "decorated stretch"})
292    if soup is None:
293        raise ParseError("Error in parsing messages.")
294    tbody = soup.find("tbody")
295    if not isinstance(tbody, Tag):
296        raise ParseError("Error in parsing messages (tbody).")
297    tds = tbody.find_all("tr", attrs={"class": ["line0", "line1"]})
298    if tds[0].text.strip() == "Brak wiadomości":
299        return []
300    for td in tds:
301        unread = False
302        hasAttachment = False
303        message_data: List[Tag] = td.find_all("td")
304        if len(message_data) < 6:
305            raise ParseError("Message data has less than 6 elements")
306        _tick, attachment, author, title, date, _trash = message_data[:6]
307        if attachment.find("img"):
308            hasAttachment = True
309        style = title.get("style")
310        if not isinstance(style, List) and not isinstance(style, str):
311            style = []
312        if "font-weight: bold" in style:
313            unread = True
314
315        author_a = author.find("a")
316        href = ""
317        if isinstance(author_a, Tag):
318            href = author_a.attrs.get("href", "")
319            href = _sanitize_href(href)
320
321        author = author.text
322        title = title.text
323        date = date.text
324        m = Message(author, title, date, href, unread, hasAttachment)
325        msgs.append(m)
326    return msgs
327
328
329def get_max_page_number(client: Client) -> int:
330    """
331    Retrieves the maximum page number of messages.
332
333    Args:
334        client (Client): The client object for making HTTP requests.
335
336    Returns:
337        int: The maximum page number.
338    """
339    soup = no_access_check(BeautifulSoup(client.get(client.MESSAGE_URL).text, "lxml"))
340    try:
341        pages = soup.select_one("div.pagination > span")
342        if not pages:
343            return 0
344        max_pages = pages.text.replace("\xa0", "")
345        max_pages_re = re.search("z[0-9]*", max_pages)
346        if max_pages_re is None:
347            return 0
348        max_pages_number = int(max_pages_re.group(0).replace("z", ""))
349    except:
350        raise ParseError("Error while trying to get max page number.")
351    return max_pages_number - 1
352
353
354def get_received(client: Client, page: int) -> List[Message]:
355    """
356    Retrieves received messages from a specific page.
357
358    Args:
359        client (Client): The client object for making HTTP requests.
360        page (int): The page number of messages to retrieve.
361
362    Returns:
363        List[Message]: A list of received Message objects.
364    """
365    payload = {
366        "numer_strony105": page,
367        "porcjowanie_pojemnik105": "105",
368    }
369    response = client.post(client.MESSAGE_URL, data=payload)
370    soup = no_access_check(BeautifulSoup(response.text, "lxml"))
371    received_msgs = parse(soup)
372    return received_msgs
373
374
375def get_sent(client: Client, page: int) -> List[Message]:
376    """
377    Retrieves sent messages from a specific page.
378
379    Args:
380        client (Client): The client object for making HTTP requests.
381        page (int): The page number of messages to retrieve.
382
383    Returns:
384        List[Message]: A list of sent Message objects.
385    """
386    payload = {
387        "numer_strony105": page,
388        "porcjowanie_pojemnik105": "105",
389    }
390    response = client.post(client.SEND_MESSAGE_URL, data=payload)
391    soup = no_access_check(BeautifulSoup(response.text, "lxml"))
392    received_msgs = parse_sent(soup)
393    return received_msgs