lernsax.util.client

Communicator code to talk with LernSax.de

  1"""
  2Communicator code to talk with LernSax.de
  3"""
  4
  5from abc import ABC
  6from box import Box
  7from . import exceptions
  8
  9# Abstract ApiClient only as a skeleton
 10
 11
 12class ApiClient(ABC):
 13    def pack_responses(self, results: list, main_answer_index: int) -> dict:
 14        """
 15        Packs multiple method responses together.
 16        The main response is accessible through the "result" key.
 17        Helper method responses are accessible through the "helpers" key of the returned dict.
 18        """
 19        packed_results = Box({"result": results.pop(
 20            main_answer_index), "helpers": results})
 21        return packed_results.to_dict()
 22
 23    def jsonrpc(self, data: list):
 24        """
 25        Prepares Data to be sent to the API in correct jsonrpc format
 26        """
 27        return [{"id": k[0], "jsonrpc": "2.0", "method": k[1], "params": k[2]} for k in data]
 28
 29    async def login(self, email: str = "", password: str = "") -> dict:
 30        """ Enter the LernSax session """
 31        if not email or not password:
 32            email, password = self.email, self.password
 33        results_raw = await self.post(
 34            self.jsonrpc(
 35                [
 36                    [
 37                        1,
 38                        "login",
 39                        {"login": email, "password": password,
 40                            "get_miniature": True},
 41                    ],
 42                    [999, "get_information", {}],
 43                ]
 44            )
 45        )
 46        results = [Box(res) for res in results_raw]
 47
 48        if not results[0].result["return"] == "OK":
 49            raise exceptions.error_handler(
 50                results[0].result.errno)(results_raw[0]["result"])
 51
 52        self.sid, self.email, self.password, self.member_of = (
 53            results[1].result.session_id,
 54            email,
 55            password,
 56            [member.login for member in results[0].result.member],
 57        )
 58        return self.pack_responses(results_raw, 0)
 59
 60    async def refresh_session(self) -> dict:
 61        """ Refreshes current LernSax session. """
 62        if not self.sid:
 63            raise exceptions.NotLoggedIn()
 64        results_raw = await self.post(self.jsonrpc([[1, "set_session", {"session_id": self.sid}]]))
 65        return self.pack_responses(results_raw, 0)
 66
 67    async def logout(self) -> dict:
 68        """ Exit the LernSax session """
 69        results_raw = await self.post(
 70            self.jsonrpc(
 71                [
 72                    [1, "set_session", {"session_id": self.sid}],
 73                    [2, "set_focus", {"object": "settings"}],
 74                    [3, "logout", {}],
 75                ]
 76            )
 77        )
 78        results = [Box(res) for res in results_raw]
 79        if not results[-1].result["return"] == "OK":
 80            raise exceptions.error_handler(
 81                results[-1].result.errno)(results_raw[-1]["result"])
 82        self.sid = ""
 83        return self.pack_responses(results_raw, 2)
 84
 85    async def get_tasks(self, group: str) -> dict:
 86        """ Get LernSax tasks, thanks to  TKFRvisionOfficial for finding the json rpc request """
 87        if not self.sid:
 88            raise exceptions.NotLoggedIn()
 89        results_raw = await self.post(
 90            self.jsonrpc(
 91                [
 92                    [1, "set_session", {"session_id": self.sid}],
 93                    [2, "set_focus", {"login": group, "object": "tasks"}],
 94                    [3, "get_entries", {}],
 95                ]
 96            )
 97        )
 98        return self.pack_responses(results_raw, 2)
 99
100    # FileRequest
101
102    async def get_download_url(self, login: str, id: str) -> dict:
103        """ Gets download id with the file id """
104        if not self.sid:
105            raise exceptions.NotLoggedIn()
106        results_raw = await self.post(
107            self.jsonrpc(
108                [
109                    [1, "set_session", {"session_id": self.sid}],
110                    [2, "set_focus", {"login": login, "object": "files"}],
111                    [3, "get_file_download_url", {"id": id}],
112                ]
113            )
114        )
115        return self.pack_responses(results_raw, 2)
116
117    # ForumRequest
118
119    async def get_board(self, login: str) -> dict:
120        """ Gets messages board for specified login """
121        if not self.sid:
122            raise exceptions.NotLoggedIn()
123        results_raw = await self.post(
124            self.jsonrpc(
125                [
126                    [1, "set_session", {"session_id": self.sid}],
127                    [2, "set_focus", {"login": login, "object": "files"}],
128                    [3, "get_entries", {}],
129                ]
130            )
131        )
132        return self.pack_responses(results_raw, 2)
133
134    async def add_board_entry(self, login: str, title: str, text: str, color: str) -> dict:
135        """Adds board entry for specified (group-)login.
136        color must be a hexadecimal color code
137        """
138        if not self.sid:
139            raise exceptions.NotLoggedIn()
140        results_raw = await self.post(
141            self.jsonrpc(
142                [
143                    [1, "set_session", {"session_id": self.sid}],
144                    [2, "set_focus", {"login": login, "object": "board"}],
145                    [
146                        3,
147                        "add_entry",
148                        {"title": title, "text": text, "color": color},
149                    ],
150                ]
151            )
152        )
153        results = [Box(res) for res in results_raw]
154        if not results[-1].result["return"] == "OK":
155            raise exceptions.error_handler(
156                results[-1].result.errno)(results_raw[-1]["result"])
157        return self.pack_responses(results_raw, 2)
158
159    # NotesRequest
160
161    async def get_notes(self, login: str) -> dict:
162        """ Gets notes for specified login """
163        if not self.sid:
164            raise exceptions.NotLoggedIn()
165        results_raw = await self.post(
166            self.jsonrpc(
167                [
168                    [1, "set_session", {"session_id": self.sid}],
169                    [2, "set_focus", {"login": login, "object": "notes"}],
170                    [3, "get_entries", {}],
171                ]
172            )
173        )
174        return self.pack_responses(results_raw, 2)
175
176    async def add_note(self, title: str, text: str) -> dict:
177        """ adds a note """
178        if not self.sid:
179            raise exceptions.NotLoggedIn()
180        results_raw = await self.post(
181            self.jsonrpc(
182                [
183                    [1, "set_session", {"session_id": self.sid}],
184                    [2, "set_focus", {"object": "notes"}],
185                    [3, "add_entry", {"text": text, "title": title}],
186                ]
187            )
188        )
189        results = [Box(res) for res in results_raw]
190        if not results[-1].result["return"] == "OK":
191            raise exceptions.error_handler(
192                results[-1].result.errno)(results_raw[-1]["result"])
193        return self.pack_responses(results_raw, 2)
194
195    async def delete_note(self, id: str) -> dict:
196        """ deletes a note """
197        if not self.sid:
198            raise exceptions.NotLoggedIn()
199        results_raw = await self.post(
200            self.jsonrpc(
201                [
202                    [1, "set_session", {"session_id": self.sid}],
203                    [2, "set_focus", {"object": "notes"}],
204                    [3, "delete_entry", {"id": id}],
205                ]
206            )
207        )
208        results = [Box(res) for res in results_raw]
209        if not results[-1].result["return"] == "OK":
210            raise exceptions.error_handler(
211                results[-1].result.errno)(results_raw[-1]["result"])
212        return self.pack_responses(results_raw, 2)
213
214    #  EmailRequest
215
216    async def send_email(self, to: str, subject: str, body: str) -> dict:
217        """ Sends an email """
218        if not self.sid:
219            raise exceptions.NotLoggedIn()
220        results_raw = await self.post(
221            self.jsonrpc(
222                [
223                    [1, "set_session", {"session_id": self.sid}],
224                    [2, "set_focus", {"object": "mailbox"}],
225                    [3, "send_mail", {
226                        "to": to, "subject": subject, "body_plain": body}],
227                ]
228            )
229        )
230        results = [Box(res) for res in results_raw]
231        if not results[-1].result["return"] == "OK":
232            raise exceptions.error_handler(
233                results[-1].result.errno)(results_raw[-1]["result"])
234        return self.pack_responses(results_raw, 2)
235
236    async def get_emails(self, folder_id: str) -> dict:
237        """ Gets emails from a folder id """
238        if not self.sid:
239            raise exceptions.NotLoggedIn()
240        results_raw = await self.post(
241            self.jsonrpc(
242                [
243                    [1, "set_session", {"session_id": self.sid}],
244                    [2, "set_focus", {"object": "mailbox"}],
245                    [3, "get_messages", {"folder_id": folder_id}],
246                ]
247            )
248        )
249        return self.pack_responses(results_raw, 2)
250
251    async def read_email(self, folder_id: str, message_id: int) -> dict:
252        """ reads an email with a certain message id """
253        if not self.sid:
254            raise exceptions.NotLoggedIn()
255        results_raw = await self.post(
256            self.jsonrpc(
257                [
258                    [1, "set_session", {"session_id": self.sid}],
259                    [2, "set_focus", {"object": "mailbox"}],
260                    [3, "read_message", {
261                        "folder_id": folder_id, "message_id": message_id}],
262                ]
263            )
264        )
265        results = [Box(res) for res in results_raw]
266        if not results[-1].result["return"] == "OK":
267            raise exceptions.error_handler(
268                results[-1].result.errno)(results_raw[-1])["result"]
269        return self.pack_responses(results_raw, 2)
270
271    async def get_email_folders(self):
272        """ returns the folders to get the id """
273        if not self.sid:
274            raise exceptions.NotLoggedIn()
275        results_raw = await self.post(
276            self.jsonrpc(
277                [
278                    [1, "set_session", {"session_id": self.sid}],
279                    [2, "set_focus", {"object": "mailbox"}],
280                    [3, "get_folders", {}],
281                ]
282            )
283        )
284
285        return self.pack_responses(results_raw, 2)
286
287    # MessengerRequest
288
289    async def read_quickmessages(self) -> dict:
290        """ returns quickmessages """
291        if not self.sid:
292            raise exceptions.NotLoggedIn()
293        results_raw = await self.post(
294            self.jsonrpc(
295                [
296                    [1, "set_session", {"session_id": self.sid}],
297                    [2, "set_focus", {"object": "messenger"}],
298                    [3, "read_quick_messages", {"export_session_file": 0}],
299                ]
300            )
301        )
302        return self.pack_responses(results_raw, 2)
303
304    async def send_quickmessage(self, login: str, text: str) -> dict:
305        """ Sends a quickmessage to an email holder """
306        if not self.sid:
307            raise exceptions.NotLoggedIn()
308        results_raw = await self.post(
309            self.jsonrpc(
310                [
311                    [1, "set_session", {"session_id": self.sid}],
312                    [2, "set_focus", {"object": "messenger"}],
313                    [3, "send_quick_message", {
314                        "login": login, "text": text, "import_session_file": 0}],
315                ]
316            )
317        )
318        results = [Box(res) for res in results_raw]
319        if not results[-1].result["return"] == "OK":
320            raise exceptions.error_handler(
321                results[-1].result.errno)(results_raw[-1]["result"])
322        return self.pack_responses(results_raw, 2)
323
324    async def get_quickmessage_history(self, start_id: int) -> dict:
325        """ get quickmessage history """
326        if not self.sid:
327            raise exceptions.NotLoggedIn()
328        results_raw = await self.post(
329            self.jsonrpc(
330                [
331                    [1, "set_session", {"session_id": self.sid}],
332                    [2, "set_focus", {"object": "messenger"}],
333                    [3, "get_history", {
334                        "start_id": start_id, "export_session_file": 0}],
335                ]
336            )
337        )
338        results = [Box(res) for res in results_raw]
339        if (not results[-1].result["return"] == "OK") and (results[-1].result.errno == "107" or results[-1].result.errno == "103"):
340            raise exceptions.error_handler(
341                results[-1].result.errno)(results_raw[-1]["result"])
342        return self.pack_responses(results_raw, 2)
343
344    async def group_lernsax_quickmessage_history_by_chat(self, quickmsg_history: list):
345        """Groups LernSax quickmessage history by chat email and date.
346        The returned LernSax quickmessage history only includes a list of all messages. They are not grouped by chat emails yet.
347        This function will group all quickmessages for same chat emails together.
348        In the returned dict the messages associated to a chat are sorted by the date they were sent.
349        Parse the returned data from get_lernsax_quickmessage_history() as quickmsg_history attr.
350        """
351        messages = quickmsg_history["result"]["result"]["messages"]
352        grouped_messages = Box({})
353        for msg in messages:
354            msg = Box(msg)
355            msg.date = int(msg.date)
356            receiving_chat_email = msg.to.login
357            receiving_chat_name = msg.to.name_hr
358            receiving_chat_type = msg.to.type
359            if not receiving_chat_email in grouped_messages:
360                grouped_messages[receiving_chat_email] = {
361                    "chat_name": receiving_chat_name,
362                    "chat_type": receiving_chat_type,
363                    "messages": [],
364                }
365            new_message = {
366                "id": msg.id,
367                "text": msg.text,
368                "date": msg.date,
369                "flags": msg.flags,
370            }
371            if (
372                len(grouped_messages[receiving_chat_email].messages) == 0
373                or msg.date >= grouped_messages[receiving_chat_email].messages[-1].date
374            ):
375                grouped_messages[receiving_chat_email].messages.append(
376                    new_message)
377            else:
378                # By async default the messages in the quickmessage history should be sorted by date. If there is a mistake in the sorting
379                # those statements are called.
380                current_index = 0
381                for existing_msg in grouped_messages[receiving_chat_email].messages:
382                    if existing_msg.date >= msg.date:
383                        grouped_messages[receiving_chat_email].messages.insert(
384                            current_index, new_message)
385                        break
386
387                    current_index += 1
388        return grouped_messages.to_dict()
class ApiClient(abc.ABC): 
 13class ApiClient(ABC):
 14    def pack_responses(self, results: list, main_answer_index: int) -> dict:
 15        """
 16        Packs multiple method responses together.
 17        The main response is accessible through the "result" key.
 18        Helper method responses are accessible through the "helpers" key of the returned dict.
 19        """
 20        packed_results = Box({"result": results.pop(
 21            main_answer_index), "helpers": results})
 22        return packed_results.to_dict()
 23
 24    def jsonrpc(self, data: list):
 25        """
 26        Prepares Data to be sent to the API in correct jsonrpc format
 27        """
 28        return [{"id": k[0], "jsonrpc": "2.0", "method": k[1], "params": k[2]} for k in data]
 29
 30    async def login(self, email: str = "", password: str = "") -> dict:
 31        """ Enter the LernSax session """
 32        if not email or not password:
 33            email, password = self.email, self.password
 34        results_raw = await self.post(
 35            self.jsonrpc(
 36                [
 37                    [
 38                        1,
 39                        "login",
 40                        {"login": email, "password": password,
 41                            "get_miniature": True},
 42                    ],
 43                    [999, "get_information", {}],
 44                ]
 45            )
 46        )
 47        results = [Box(res) for res in results_raw]
 48
 49        if not results[0].result["return"] == "OK":
 50            raise exceptions.error_handler(
 51                results[0].result.errno)(results_raw[0]["result"])
 52
 53        self.sid, self.email, self.password, self.member_of = (
 54            results[1].result.session_id,
 55            email,
 56            password,
 57            [member.login for member in results[0].result.member],
 58        )
 59        return self.pack_responses(results_raw, 0)
 60
 61    async def refresh_session(self) -> dict:
 62        """ Refreshes current LernSax session. """
 63        if not self.sid:
 64            raise exceptions.NotLoggedIn()
 65        results_raw = await self.post(self.jsonrpc([[1, "set_session", {"session_id": self.sid}]]))
 66        return self.pack_responses(results_raw, 0)
 67
 68    async def logout(self) -> dict:
 69        """ Exit the LernSax session """
 70        results_raw = await self.post(
 71            self.jsonrpc(
 72                [
 73                    [1, "set_session", {"session_id": self.sid}],
 74                    [2, "set_focus", {"object": "settings"}],
 75                    [3, "logout", {}],
 76                ]
 77            )
 78        )
 79        results = [Box(res) for res in results_raw]
 80        if not results[-1].result["return"] == "OK":
 81            raise exceptions.error_handler(
 82                results[-1].result.errno)(results_raw[-1]["result"])
 83        self.sid = ""
 84        return self.pack_responses(results_raw, 2)
 85
 86    async def get_tasks(self, group: str) -> dict:
 87        """ Get LernSax tasks, thanks to  TKFRvisionOfficial for finding the json rpc request """
 88        if not self.sid:
 89            raise exceptions.NotLoggedIn()
 90        results_raw = await self.post(
 91            self.jsonrpc(
 92                [
 93                    [1, "set_session", {"session_id": self.sid}],
 94                    [2, "set_focus", {"login": group, "object": "tasks"}],
 95                    [3, "get_entries", {}],
 96                ]
 97            )
 98        )
 99        return self.pack_responses(results_raw, 2)
100
101    # FileRequest
102
103    async def get_download_url(self, login: str, id: str) -> dict:
104        """ Gets download id with the file id """
105        if not self.sid:
106            raise exceptions.NotLoggedIn()
107        results_raw = await self.post(
108            self.jsonrpc(
109                [
110                    [1, "set_session", {"session_id": self.sid}],
111                    [2, "set_focus", {"login": login, "object": "files"}],
112                    [3, "get_file_download_url", {"id": id}],
113                ]
114            )
115        )
116        return self.pack_responses(results_raw, 2)
117
118    # ForumRequest
119
120    async def get_board(self, login: str) -> dict:
121        """ Gets messages board for specified login """
122        if not self.sid:
123            raise exceptions.NotLoggedIn()
124        results_raw = await self.post(
125            self.jsonrpc(
126                [
127                    [1, "set_session", {"session_id": self.sid}],
128                    [2, "set_focus", {"login": login, "object": "files"}],
129                    [3, "get_entries", {}],
130                ]
131            )
132        )
133        return self.pack_responses(results_raw, 2)
134
135    async def add_board_entry(self, login: str, title: str, text: str, color: str) -> dict:
136        """Adds board entry for specified (group-)login.
137        color must be a hexadecimal color code
138        """
139        if not self.sid:
140            raise exceptions.NotLoggedIn()
141        results_raw = await self.post(
142            self.jsonrpc(
143                [
144                    [1, "set_session", {"session_id": self.sid}],
145                    [2, "set_focus", {"login": login, "object": "board"}],
146                    [
147                        3,
148                        "add_entry",
149                        {"title": title, "text": text, "color": color},
150                    ],
151                ]
152            )
153        )
154        results = [Box(res) for res in results_raw]
155        if not results[-1].result["return"] == "OK":
156            raise exceptions.error_handler(
157                results[-1].result.errno)(results_raw[-1]["result"])
158        return self.pack_responses(results_raw, 2)
159
160    # NotesRequest
161
162    async def get_notes(self, login: str) -> dict:
163        """ Gets notes for specified login """
164        if not self.sid:
165            raise exceptions.NotLoggedIn()
166        results_raw = await self.post(
167            self.jsonrpc(
168                [
169                    [1, "set_session", {"session_id": self.sid}],
170                    [2, "set_focus", {"login": login, "object": "notes"}],
171                    [3, "get_entries", {}],
172                ]
173            )
174        )
175        return self.pack_responses(results_raw, 2)
176
177    async def add_note(self, title: str, text: str) -> dict:
178        """ adds a note """
179        if not self.sid:
180            raise exceptions.NotLoggedIn()
181        results_raw = await self.post(
182            self.jsonrpc(
183                [
184                    [1, "set_session", {"session_id": self.sid}],
185                    [2, "set_focus", {"object": "notes"}],
186                    [3, "add_entry", {"text": text, "title": title}],
187                ]
188            )
189        )
190        results = [Box(res) for res in results_raw]
191        if not results[-1].result["return"] == "OK":
192            raise exceptions.error_handler(
193                results[-1].result.errno)(results_raw[-1]["result"])
194        return self.pack_responses(results_raw, 2)
195
196    async def delete_note(self, id: str) -> dict:
197        """ deletes a note """
198        if not self.sid:
199            raise exceptions.NotLoggedIn()
200        results_raw = await self.post(
201            self.jsonrpc(
202                [
203                    [1, "set_session", {"session_id": self.sid}],
204                    [2, "set_focus", {"object": "notes"}],
205                    [3, "delete_entry", {"id": id}],
206                ]
207            )
208        )
209        results = [Box(res) for res in results_raw]
210        if not results[-1].result["return"] == "OK":
211            raise exceptions.error_handler(
212                results[-1].result.errno)(results_raw[-1]["result"])
213        return self.pack_responses(results_raw, 2)
214
215    #  EmailRequest
216
217    async def send_email(self, to: str, subject: str, body: str) -> dict:
218        """ Sends an email """
219        if not self.sid:
220            raise exceptions.NotLoggedIn()
221        results_raw = await self.post(
222            self.jsonrpc(
223                [
224                    [1, "set_session", {"session_id": self.sid}],
225                    [2, "set_focus", {"object": "mailbox"}],
226                    [3, "send_mail", {
227                        "to": to, "subject": subject, "body_plain": body}],
228                ]
229            )
230        )
231        results = [Box(res) for res in results_raw]
232        if not results[-1].result["return"] == "OK":
233            raise exceptions.error_handler(
234                results[-1].result.errno)(results_raw[-1]["result"])
235        return self.pack_responses(results_raw, 2)
236
237    async def get_emails(self, folder_id: str) -> dict:
238        """ Gets emails from a folder id """
239        if not self.sid:
240            raise exceptions.NotLoggedIn()
241        results_raw = await self.post(
242            self.jsonrpc(
243                [
244                    [1, "set_session", {"session_id": self.sid}],
245                    [2, "set_focus", {"object": "mailbox"}],
246                    [3, "get_messages", {"folder_id": folder_id}],
247                ]
248            )
249        )
250        return self.pack_responses(results_raw, 2)
251
252    async def read_email(self, folder_id: str, message_id: int) -> dict:
253        """ reads an email with a certain message id """
254        if not self.sid:
255            raise exceptions.NotLoggedIn()
256        results_raw = await self.post(
257            self.jsonrpc(
258                [
259                    [1, "set_session", {"session_id": self.sid}],
260                    [2, "set_focus", {"object": "mailbox"}],
261                    [3, "read_message", {
262                        "folder_id": folder_id, "message_id": message_id}],
263                ]
264            )
265        )
266        results = [Box(res) for res in results_raw]
267        if not results[-1].result["return"] == "OK":
268            raise exceptions.error_handler(
269                results[-1].result.errno)(results_raw[-1])["result"]
270        return self.pack_responses(results_raw, 2)
271
272    async def get_email_folders(self):
273        """ returns the folders to get the id """
274        if not self.sid:
275            raise exceptions.NotLoggedIn()
276        results_raw = await self.post(
277            self.jsonrpc(
278                [
279                    [1, "set_session", {"session_id": self.sid}],
280                    [2, "set_focus", {"object": "mailbox"}],
281                    [3, "get_folders", {}],
282                ]
283            )
284        )
285
286        return self.pack_responses(results_raw, 2)
287
288    # MessengerRequest
289
290    async def read_quickmessages(self) -> dict:
291        """ returns quickmessages """
292        if not self.sid:
293            raise exceptions.NotLoggedIn()
294        results_raw = await self.post(
295            self.jsonrpc(
296                [
297                    [1, "set_session", {"session_id": self.sid}],
298                    [2, "set_focus", {"object": "messenger"}],
299                    [3, "read_quick_messages", {"export_session_file": 0}],
300                ]
301            )
302        )
303        return self.pack_responses(results_raw, 2)
304
305    async def send_quickmessage(self, login: str, text: str) -> dict:
306        """ Sends a quickmessage to an email holder """
307        if not self.sid:
308            raise exceptions.NotLoggedIn()
309        results_raw = await self.post(
310            self.jsonrpc(
311                [
312                    [1, "set_session", {"session_id": self.sid}],
313                    [2, "set_focus", {"object": "messenger"}],
314                    [3, "send_quick_message", {
315                        "login": login, "text": text, "import_session_file": 0}],
316                ]
317            )
318        )
319        results = [Box(res) for res in results_raw]
320        if not results[-1].result["return"] == "OK":
321            raise exceptions.error_handler(
322                results[-1].result.errno)(results_raw[-1]["result"])
323        return self.pack_responses(results_raw, 2)
324
325    async def get_quickmessage_history(self, start_id: int) -> dict:
326        """ get quickmessage history """
327        if not self.sid:
328            raise exceptions.NotLoggedIn()
329        results_raw = await self.post(
330            self.jsonrpc(
331                [
332                    [1, "set_session", {"session_id": self.sid}],
333                    [2, "set_focus", {"object": "messenger"}],
334                    [3, "get_history", {
335                        "start_id": start_id, "export_session_file": 0}],
336                ]
337            )
338        )
339        results = [Box(res) for res in results_raw]
340        if (not results[-1].result["return"] == "OK") and (results[-1].result.errno == "107" or results[-1].result.errno == "103"):
341            raise exceptions.error_handler(
342                results[-1].result.errno)(results_raw[-1]["result"])
343        return self.pack_responses(results_raw, 2)
344
345    async def group_lernsax_quickmessage_history_by_chat(self, quickmsg_history: list):
346        """Groups LernSax quickmessage history by chat email and date.
347        The returned LernSax quickmessage history only includes a list of all messages. They are not grouped by chat emails yet.
348        This function will group all quickmessages for same chat emails together.
349        In the returned dict the messages associated to a chat are sorted by the date they were sent.
350        Parse the returned data from get_lernsax_quickmessage_history() as quickmsg_history attr.
351        """
352        messages = quickmsg_history["result"]["result"]["messages"]
353        grouped_messages = Box({})
354        for msg in messages:
355            msg = Box(msg)
356            msg.date = int(msg.date)
357            receiving_chat_email = msg.to.login
358            receiving_chat_name = msg.to.name_hr
359            receiving_chat_type = msg.to.type
360            if not receiving_chat_email in grouped_messages:
361                grouped_messages[receiving_chat_email] = {
362                    "chat_name": receiving_chat_name,
363                    "chat_type": receiving_chat_type,
364                    "messages": [],
365                }
366            new_message = {
367                "id": msg.id,
368                "text": msg.text,
369                "date": msg.date,
370                "flags": msg.flags,
371            }
372            if (
373                len(grouped_messages[receiving_chat_email].messages) == 0
374                or msg.date >= grouped_messages[receiving_chat_email].messages[-1].date
375            ):
376                grouped_messages[receiving_chat_email].messages.append(
377                    new_message)
378            else:
379                # By async default the messages in the quickmessage history should be sorted by date. If there is a mistake in the sorting
380                # those statements are called.
381                current_index = 0
382                for existing_msg in grouped_messages[receiving_chat_email].messages:
383                    if existing_msg.date >= msg.date:
384                        grouped_messages[receiving_chat_email].messages.insert(
385                            current_index, new_message)
386                        break
387
388                    current_index += 1
389        return grouped_messages.to_dict()

Helper class that provides a standard way to create an ABC using inheritance.

ApiClient()
def pack_responses(self, results: list, main_answer_index: int) -> dict: 
14    def pack_responses(self, results: list, main_answer_index: int) -> dict:
15        """
16        Packs multiple method responses together.
17        The main response is accessible through the "result" key.
18        Helper method responses are accessible through the "helpers" key of the returned dict.
19        """
20        packed_results = Box({"result": results.pop(
21            main_answer_index), "helpers": results})
22        return packed_results.to_dict()

Packs multiple method responses together. The main response is accessible through the "result" key. Helper method responses are accessible through the "helpers" key of the returned dict.

def jsonrpc(self, data: list) 
24    def jsonrpc(self, data: list):
25        """
26        Prepares Data to be sent to the API in correct jsonrpc format
27        """
28        return [{"id": k[0], "jsonrpc": "2.0", "method": k[1], "params": k[2]} for k in data]

Prepares Data to be sent to the API in correct jsonrpc format

async def login(self, email: str = '', password: str = '') -> dict: 
30    async def login(self, email: str = "", password: str = "") -> dict:
31        """ Enter the LernSax session """
32        if not email or not password:
33            email, password = self.email, self.password
34        results_raw = await self.post(
35            self.jsonrpc(
36                [
37                    [
38                        1,
39                        "login",
40                        {"login": email, "password": password,
41                            "get_miniature": True},
42                    ],
43                    [999, "get_information", {}],
44                ]
45            )
46        )
47        results = [Box(res) for res in results_raw]
48
49        if not results[0].result["return"] == "OK":
50            raise exceptions.error_handler(
51                results[0].result.errno)(results_raw[0]["result"])
52
53        self.sid, self.email, self.password, self.member_of = (
54            results[1].result.session_id,
55            email,
56            password,
57            [member.login for member in results[0].result.member],
58        )
59        return self.pack_responses(results_raw, 0)

Enter the LernSax session

async def refresh_session(self) -> dict: 
61    async def refresh_session(self) -> dict:
62        """ Refreshes current LernSax session. """
63        if not self.sid:
64            raise exceptions.NotLoggedIn()
65        results_raw = await self.post(self.jsonrpc([[1, "set_session", {"session_id": self.sid}]]))
66        return self.pack_responses(results_raw, 0)

Refreshes current LernSax session.

async def logout(self) -> dict: 
68    async def logout(self) -> dict:
69        """ Exit the LernSax session """
70        results_raw = await self.post(
71            self.jsonrpc(
72                [
73                    [1, "set_session", {"session_id": self.sid}],
74                    [2, "set_focus", {"object": "settings"}],
75                    [3, "logout", {}],
76                ]
77            )
78        )
79        results = [Box(res) for res in results_raw]
80        if not results[-1].result["return"] == "OK":
81            raise exceptions.error_handler(
82                results[-1].result.errno)(results_raw[-1]["result"])
83        self.sid = ""
84        return self.pack_responses(results_raw, 2)

Exit the LernSax session

async def get_tasks(self, group: str) -> dict: 
86    async def get_tasks(self, group: str) -> dict:
87        """ Get LernSax tasks, thanks to  TKFRvisionOfficial for finding the json rpc request """
88        if not self.sid:
89            raise exceptions.NotLoggedIn()
90        results_raw = await self.post(
91            self.jsonrpc(
92                [
93                    [1, "set_session", {"session_id": self.sid}],
94                    [2, "set_focus", {"login": group, "object": "tasks"}],
95                    [3, "get_entries", {}],
96                ]
97            )
98        )
99        return self.pack_responses(results_raw, 2)

Get LernSax tasks, thanks to TKFRvisionOfficial for finding the json rpc request

async def get_download_url(self, login: str, id: str) -> dict: 
103    async def get_download_url(self, login: str, id: str) -> dict:
104        """ Gets download id with the file id """
105        if not self.sid:
106            raise exceptions.NotLoggedIn()
107        results_raw = await self.post(
108            self.jsonrpc(
109                [
110                    [1, "set_session", {"session_id": self.sid}],
111                    [2, "set_focus", {"login": login, "object": "files"}],
112                    [3, "get_file_download_url", {"id": id}],
113                ]
114            )
115        )
116        return self.pack_responses(results_raw, 2)

Gets download id with the file id

async def get_board(self, login: str) -> dict: 
120    async def get_board(self, login: str) -> dict:
121        """ Gets messages board for specified login """
122        if not self.sid:
123            raise exceptions.NotLoggedIn()
124        results_raw = await self.post(
125            self.jsonrpc(
126                [
127                    [1, "set_session", {"session_id": self.sid}],
128                    [2, "set_focus", {"login": login, "object": "files"}],
129                    [3, "get_entries", {}],
130                ]
131            )
132        )
133        return self.pack_responses(results_raw, 2)

Gets messages board for specified login

async def add_board_entry(self, login: str, title: str, text: str, color: str) -> dict: 
135    async def add_board_entry(self, login: str, title: str, text: str, color: str) -> dict:
136        """Adds board entry for specified (group-)login.
137        color must be a hexadecimal color code
138        """
139        if not self.sid:
140            raise exceptions.NotLoggedIn()
141        results_raw = await self.post(
142            self.jsonrpc(
143                [
144                    [1, "set_session", {"session_id": self.sid}],
145                    [2, "set_focus", {"login": login, "object": "board"}],
146                    [
147                        3,
148                        "add_entry",
149                        {"title": title, "text": text, "color": color},
150                    ],
151                ]
152            )
153        )
154        results = [Box(res) for res in results_raw]
155        if not results[-1].result["return"] == "OK":
156            raise exceptions.error_handler(
157                results[-1].result.errno)(results_raw[-1]["result"])
158        return self.pack_responses(results_raw, 2)

Adds board entry for specified (group-)login. color must be a hexadecimal color code

async def get_notes(self, login: str) -> dict: 
162    async def get_notes(self, login: str) -> dict:
163        """ Gets notes for specified login """
164        if not self.sid:
165            raise exceptions.NotLoggedIn()
166        results_raw = await self.post(
167            self.jsonrpc(
168                [
169                    [1, "set_session", {"session_id": self.sid}],
170                    [2, "set_focus", {"login": login, "object": "notes"}],
171                    [3, "get_entries", {}],
172                ]
173            )
174        )
175        return self.pack_responses(results_raw, 2)

Gets notes for specified login

async def add_note(self, title: str, text: str) -> dict: 
177    async def add_note(self, title: str, text: str) -> dict:
178        """ adds a note """
179        if not self.sid:
180            raise exceptions.NotLoggedIn()
181        results_raw = await self.post(
182            self.jsonrpc(
183                [
184                    [1, "set_session", {"session_id": self.sid}],
185                    [2, "set_focus", {"object": "notes"}],
186                    [3, "add_entry", {"text": text, "title": title}],
187                ]
188            )
189        )
190        results = [Box(res) for res in results_raw]
191        if not results[-1].result["return"] == "OK":
192            raise exceptions.error_handler(
193                results[-1].result.errno)(results_raw[-1]["result"])
194        return self.pack_responses(results_raw, 2)

adds a note

async def delete_note(self, id: str) -> dict: 
196    async def delete_note(self, id: str) -> dict:
197        """ deletes a note """
198        if not self.sid:
199            raise exceptions.NotLoggedIn()
200        results_raw = await self.post(
201            self.jsonrpc(
202                [
203                    [1, "set_session", {"session_id": self.sid}],
204                    [2, "set_focus", {"object": "notes"}],
205                    [3, "delete_entry", {"id": id}],
206                ]
207            )
208        )
209        results = [Box(res) for res in results_raw]
210        if not results[-1].result["return"] == "OK":
211            raise exceptions.error_handler(
212                results[-1].result.errno)(results_raw[-1]["result"])
213        return self.pack_responses(results_raw, 2)

deletes a note

async def send_email(self, to: str, subject: str, body: str) -> dict: 
217    async def send_email(self, to: str, subject: str, body: str) -> dict:
218        """ Sends an email """
219        if not self.sid:
220            raise exceptions.NotLoggedIn()
221        results_raw = await self.post(
222            self.jsonrpc(
223                [
224                    [1, "set_session", {"session_id": self.sid}],
225                    [2, "set_focus", {"object": "mailbox"}],
226                    [3, "send_mail", {
227                        "to": to, "subject": subject, "body_plain": body}],
228                ]
229            )
230        )
231        results = [Box(res) for res in results_raw]
232        if not results[-1].result["return"] == "OK":
233            raise exceptions.error_handler(
234                results[-1].result.errno)(results_raw[-1]["result"])
235        return self.pack_responses(results_raw, 2)

Sends an email

async def get_emails(self, folder_id: str) -> dict: 
237    async def get_emails(self, folder_id: str) -> dict:
238        """ Gets emails from a folder id """
239        if not self.sid:
240            raise exceptions.NotLoggedIn()
241        results_raw = await self.post(
242            self.jsonrpc(
243                [
244                    [1, "set_session", {"session_id": self.sid}],
245                    [2, "set_focus", {"object": "mailbox"}],
246                    [3, "get_messages", {"folder_id": folder_id}],
247                ]
248            )
249        )
250        return self.pack_responses(results_raw, 2)

Gets emails from a folder id

async def read_email(self, folder_id: str, message_id: int) -> dict: 
252    async def read_email(self, folder_id: str, message_id: int) -> dict:
253        """ reads an email with a certain message id """
254        if not self.sid:
255            raise exceptions.NotLoggedIn()
256        results_raw = await self.post(
257            self.jsonrpc(
258                [
259                    [1, "set_session", {"session_id": self.sid}],
260                    [2, "set_focus", {"object": "mailbox"}],
261                    [3, "read_message", {
262                        "folder_id": folder_id, "message_id": message_id}],
263                ]
264            )
265        )
266        results = [Box(res) for res in results_raw]
267        if not results[-1].result["return"] == "OK":
268            raise exceptions.error_handler(
269                results[-1].result.errno)(results_raw[-1])["result"]
270        return self.pack_responses(results_raw, 2)

reads an email with a certain message id

async def get_email_folders(self) 
272    async def get_email_folders(self):
273        """ returns the folders to get the id """
274        if not self.sid:
275            raise exceptions.NotLoggedIn()
276        results_raw = await self.post(
277            self.jsonrpc(
278                [
279                    [1, "set_session", {"session_id": self.sid}],
280                    [2, "set_focus", {"object": "mailbox"}],
281                    [3, "get_folders", {}],
282                ]
283            )
284        )
285
286        return self.pack_responses(results_raw, 2)

returns the folders to get the id

async def read_quickmessages(self) -> dict: 
290    async def read_quickmessages(self) -> dict:
291        """ returns quickmessages """
292        if not self.sid:
293            raise exceptions.NotLoggedIn()
294        results_raw = await self.post(
295            self.jsonrpc(
296                [
297                    [1, "set_session", {"session_id": self.sid}],
298                    [2, "set_focus", {"object": "messenger"}],
299                    [3, "read_quick_messages", {"export_session_file": 0}],
300                ]
301            )
302        )
303        return self.pack_responses(results_raw, 2)

returns quickmessages

async def send_quickmessage(self, login: str, text: str) -> dict: 
305    async def send_quickmessage(self, login: str, text: str) -> dict:
306        """ Sends a quickmessage to an email holder """
307        if not self.sid:
308            raise exceptions.NotLoggedIn()
309        results_raw = await self.post(
310            self.jsonrpc(
311                [
312                    [1, "set_session", {"session_id": self.sid}],
313                    [2, "set_focus", {"object": "messenger"}],
314                    [3, "send_quick_message", {
315                        "login": login, "text": text, "import_session_file": 0}],
316                ]
317            )
318        )
319        results = [Box(res) for res in results_raw]
320        if not results[-1].result["return"] == "OK":
321            raise exceptions.error_handler(
322                results[-1].result.errno)(results_raw[-1]["result"])
323        return self.pack_responses(results_raw, 2)

Sends a quickmessage to an email holder

async def get_quickmessage_history(self, start_id: int) -> dict: 
325    async def get_quickmessage_history(self, start_id: int) -> dict:
326        """ get quickmessage history """
327        if not self.sid:
328            raise exceptions.NotLoggedIn()
329        results_raw = await self.post(
330            self.jsonrpc(
331                [
332                    [1, "set_session", {"session_id": self.sid}],
333                    [2, "set_focus", {"object": "messenger"}],
334                    [3, "get_history", {
335                        "start_id": start_id, "export_session_file": 0}],
336                ]
337            )
338        )
339        results = [Box(res) for res in results_raw]
340        if (not results[-1].result["return"] == "OK") and (results[-1].result.errno == "107" or results[-1].result.errno == "103"):
341            raise exceptions.error_handler(
342                results[-1].result.errno)(results_raw[-1]["result"])
343        return self.pack_responses(results_raw, 2)

get quickmessage history

async def group_lernsax_quickmessage_history_by_chat(self, quickmsg_history: list) 
345    async def group_lernsax_quickmessage_history_by_chat(self, quickmsg_history: list):
346        """Groups LernSax quickmessage history by chat email and date.
347        The returned LernSax quickmessage history only includes a list of all messages. They are not grouped by chat emails yet.
348        This function will group all quickmessages for same chat emails together.
349        In the returned dict the messages associated to a chat are sorted by the date they were sent.
350        Parse the returned data from get_lernsax_quickmessage_history() as quickmsg_history attr.
351        """
352        messages = quickmsg_history["result"]["result"]["messages"]
353        grouped_messages = Box({})
354        for msg in messages:
355            msg = Box(msg)
356            msg.date = int(msg.date)
357            receiving_chat_email = msg.to.login
358            receiving_chat_name = msg.to.name_hr
359            receiving_chat_type = msg.to.type
360            if not receiving_chat_email in grouped_messages:
361                grouped_messages[receiving_chat_email] = {
362                    "chat_name": receiving_chat_name,
363                    "chat_type": receiving_chat_type,
364                    "messages": [],
365                }
366            new_message = {
367                "id": msg.id,
368                "text": msg.text,
369                "date": msg.date,
370                "flags": msg.flags,
371            }
372            if (
373                len(grouped_messages[receiving_chat_email].messages) == 0
374                or msg.date >= grouped_messages[receiving_chat_email].messages[-1].date
375            ):
376                grouped_messages[receiving_chat_email].messages.append(
377                    new_message)
378            else:
379                # By async default the messages in the quickmessage history should be sorted by date. If there is a mistake in the sorting
380                # those statements are called.
381                current_index = 0
382                for existing_msg in grouped_messages[receiving_chat_email].messages:
383                    if existing_msg.date >= msg.date:
384                        grouped_messages[receiving_chat_email].messages.insert(
385                            current_index, new_message)
386                        break
387
388                    current_index += 1
389        return grouped_messages.to_dict()

Groups LernSax quickmessage history by chat email and date. The returned LernSax quickmessage history only includes a list of all messages. They are not grouped by chat emails yet. This function will group all quickmessages for same chat emails together. In the returned dict the messages associated to a chat are sorted by the date they were sent. Parse the returned data from get_lernsax_quickmessage_history() as quickmsg_history attr.