refactor: Major cleanup and code modernization (#10)

* Using logger instead of print

* Adding new script

* Fix if no proper FutureTask received

* A bit cleaner

* Adding run

* Improving by moving everything into the cache

* Adding PlanningElements

* New TODO list based on planner

* WIP

* Poetry update

* ruff reformat

* Correcting failing test

* Correcting failing test

* sonar hint

* sonar hint2

Author: svaningelgem

.gitattributes

@@ -1,2 +1,4 @@
+scripts/list_all_courses text eol=lf
+scripts/smartschool_report_on_future_tasks text eol=lf
 scripts/smartschool_report_on_results text eol=lf
 *.py text eol=lf

.gitignore

@@ -4,9 +4,11 @@
 dist/
 .cache/
 __pycache__/
+logs/
 
 *.iml
 credentials.yml
 cookies.txt
 .coverage
 coverage.json
+*.log

poetry.lock

@@ -2,7 +2,7 @@
 name = "smartschool"
 version = "0.5.0"
 description = "Unofficial API interface to the smartschool system."
-authors = ["Steven 'KaReL' Van Ingelgem <[email protected]>"]
+authors = ["Steven Van Ingelgem <[email protected]>"]
 readme = "README.md"
 homepage = "https://github.com/svaningelgem/smartschool"
 license = "GPL-3.0"

pyproject.toml

@@ -18,7 +18,7 @@ export PYTHON="/opt/miniforge3/envs/smartschool/bin/python"
   echo "RESULTS-REPORT" |& tee -a "$logfile"
   $PYTHON scripts/smartschool_report_on_results |& tee -a "$logfile"
 
-  echo "FUTURE-TASKS-REPORT" |& tee -a "$logfile"
-  $PYTHON scripts/smartschool_report_on_future_tasks |& tee -a "$logfile"
+#  echo "FUTURE-TASKS-REPORT" |& tee -a "$logfile"
+#  $PYTHON scripts/smartschool_report_on_future_tasks |& tee -a "$logfile"
 
 ) 200>/var/lock/smartschool.lock

run

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from functools import lru_cache
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from datetime import date, time
+
+
+def _email_text(hour_start: datetime | time, hour_end: datetime | time, course_name: str, task_label: str, description: str) -> str:
+    return f"date: {hour_start:%Y-%m-%d}\ntime: {hour_start:%H:%M} -> {hour_end:%H:%M}\nles: {course_name}\n\n{task_label}:\n  {description}\n"
+
+
+@lru_cache(1)
+def _next_weekday() -> date | None:
+    now = datetime.now()
+
+    if (now.hour in [12, 20] and now.isoweekday() == 3) or (now.hour in [17, 20] and now.isoweekday() != 3):  # Wednesday
+        ...  # Ok
+    else:
+        return None
+
+    next_weekday = now + timedelta(days=1)
+    while next_weekday.isoweekday() in [6, 7]:  # Saturday, Sunday
+        next_weekday += timedelta(days=1)
+
+    return next_weekday.date()

scripts/_common_tasks_stuff.py

@@ -0,0 +1,8 @@
+#!/usr/bin/env python
+from logprise import logger
+
+from smartschool import Smartschool, PathCredentials, Courses
+
+Smartschool.start(PathCredentials())
+for course in Courses():
+    logger.info(course.name)

scripts/list_all_courses

@@ -5,10 +5,7 @@ from datetime import date, datetime, timedelta
 from smartschool import Courses, FutureTasks, PathCredentials, Periods, Smartschool, SmartschoolHours, SmartschoolLessons, SmartschoolMomentInfos
 from smartschool.common import IsSaved, save, send_email
 from smartschool.objects import AgendaHour, FutureTaskOneTask
-
-session = Smartschool.start(PathCredentials())
-assert 'email_from' in session.creds.other_info
-assert 'email_to' in session.creds.other_info
+from smartschool.session import session
 
 
 def _is_equal_task(a: FutureTaskOneTask, b: FutureTaskOneTask) -> bool:
@@ -100,6 +97,10 @@ def report_tasks_for_next_day(all_tasks: dict):
 def main():
     all_tasks: dict[date, dict[str, list]] = defaultdict(lambda: defaultdict(list))
 
+    Smartschool.start(PathCredentials())
+    assert 'email_from' in session.creds.other_info
+    assert 'email_to' in session.creds.other_info
+
     for day in FutureTasks().days:
         for course in day.courses:
             assert not course.items.materials, "Please correct the model to include this information."

scripts/smartschool_report_on_future_tasks

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+from collections import defaultdict
+from datetime import date
+
+from _common_tasks_stuff import _next_weekday, _email_text
+from smartschool import PathCredentials, Smartschool
+from smartschool.common import IsSaved, save, send_email
+from smartschool.objects import FutureTaskOneTask, PlannedElement
+from smartschool.planner import PlannedElements
+from smartschool.session import session
+
+
+def _is_equal_task(a: PlannedElement, b: PlannedElement) -> bool:
+    return (
+        [c.name for c in a.courses],
+        a.period.dateTimeFrom,
+        a.period.dateTimeTo,
+        a.assignmentType.name,
+        a.name
+    ) == (
+        [c.name for c in b.courses],
+        b.period.dateTimeFrom,
+        b.period.dateTimeTo,
+        b.assignmentType.name,
+        b.name
+    )
+
+
+def process_task(task: PlannedElement, all_tasks: dict) -> None:
+    """Stores in an array + report if it's a new/updated task."""
+    next_weekday = _next_weekday()
+    is_for_next_day = next_weekday < task.period.dateTimeFrom.date() or next_weekday > task.period.dateTimeTo.date()
+
+    course_name: str = task.courses[0].name
+    hour: str = task.period.dateTimeFrom.strftime("%H:%M")
+    when: str = task.period.dateTimeFrom.strftime("%Y-%m-%d")
+
+    text = _email_text(
+        hour_start=task.period.dateTimeFrom,
+        hour_end=task.period.dateTimeTo,
+        course_name=course_name,
+        task_label=task.assignmentType.name,
+        description=task.name,
+    )
+
+    if is_for_next_day:
+        all_tasks[when][hour].append(text)
+
+    status = save(type_="todo", course_name=course_name, id_=task.assignmentType.id, data=task, is_eq=_is_equal_task)
+    if status == IsSaved.SAME:
+        return
+
+    subject = f"📑 [{when} {hour}] {course_name} ({task.name})"
+    if status != IsSaved.NEW:
+        subject = f"⚠⚠ {subject}"  # There is an update...
+
+    send_email(subject=subject, text=text, email_from=session.creds.other_info['email_from'],
+               email_to=session.creds.other_info['email_to'])
+
+
+def report_tasks_for_next_day(all_tasks: dict):
+    for dag, rest in all_tasks.items():
+        text = []
+
+        for _, tasks in sorted(rest.items(), key=lambda x: x[0]):  # type: str, list[PlannedElement]
+            for task in tasks:
+                text.append(task)
+
+        text = "\n---------------------------------------\n\n".join(text)
+
+        status = save('todo', '.email', dag, data=text, extension='txt')
+        if status == IsSaved.SAME:
+            continue  # Don't re-send if nothing changed
+
+        subject = f"🍕 Todo list {dag}"
+        if status != IsSaved.NEW:
+            subject = f"⚠⚠ {subject}"  # There is an update...
+
+        send_email(
+            subject=subject,
+            text=text, email_from=session.creds.other_info['email_from'], email_to=session.creds.other_info['email_to'])
+
+
+def main():
+    all_tasks: dict[date, dict[str, list]] = defaultdict(lambda: defaultdict(list))
+
+    Smartschool.start(PathCredentials())
+    assert 'email_from' in session.creds.other_info
+    assert 'email_to' in session.creds.other_info
+
+    for element in PlannedElements():
+        process_task(element, all_tasks)
+
+    report_tasks_for_next_day(all_tasks)
+
+
+if __name__ == '__main__':
+    main()

scripts/smartschool_report_on_planned_tasks

@@ -3,11 +3,11 @@
 from logprise import logger
 from smartschool import PathCredentials, Results, Smartschool, DownloadError
 from smartschool.common import IsSaved, save, send_email
-from smartschool.objects import Result
+from smartschool.objects import ResultWithoutDetails
 from smartschool.session import session
 
 
-def is_punten_json_the_same(previous: Result, current: Result) -> bool:
+def is_punten_json_the_same(previous: ResultWithoutDetails, current: ResultWithoutDetails) -> bool:
     if previous == current:  # No need to dig into it...
         return True
 
@@ -33,15 +33,15 @@ def is_punten_json_the_same(previous: Result, current: Result) -> bool:
         "feedback/5/text",
     ]
 
-    def _grab_sub(d: Result | dict, key: list[str]) -> dict | list | str | None:
-        if not key or d is None:
+    def _grab_sub(d: ResultWithoutDetails | dict, k: list[str]) -> dict | list | str | None:
+        if not k or d is None:
             return d
 
         obj = d
-        for piece in key:
+        for piece in k:
             if piece.isnumeric():
                 piece = int(piece)
-                if piece >= len(obj):  # Not enough inside this array >> Just assume it's an empty string
+                if piece >= len(obj):  # Not enough inside this array >> Assume it's an empty string
                     return ""
 
                 obj = obj[piece]
@@ -58,17 +58,17 @@ def is_punten_json_the_same(previous: Result, current: Result) -> bool:
             if prev != curr:
                 return False
     except Exception as ex:
-        print(f"Exception received: {ex.__class__.__name__} ({ex})")
-        print("Current: ", current)
-        print("Previous: ", previous)
+        logger.error("Current: ", current)
+        logger.error("Previous: ", previous)
+        logger.exception(ex)
         raise
 
     return True
 
 
 def build_text(
     is_update: bool,
-    result: Result,
+    result: ResultWithoutDetails,
 ) -> tuple[str, str]:
     course_name = result.courses[0].name
     teacher_names = [teacher.name.startingWithFirstName for teacher in result.courses[0].teachers]
@@ -110,29 +110,29 @@ def build_text(
         0.5: "🌫",
     }
 
+    icon = "😡"
     for min_pct, symbol in limits.items():
         if result.graphic.percentage >= min_pct:
             icon = symbol
             break
-    else:
-        icon = "😡"
 
     email_subject = f"{icon} {email_subject}"
 
     return text, email_subject
 
 
-def process_result(result: Result) -> None:
-    logger.info("Processing {}", result.name)
-
+def process_result(result: ResultWithoutDetails) -> None:
     assert len(result.courses) == 1, f"Multiple courses? {result.courses}"
 
+    logger.info("[{}] Processing '{}'", result.courses[0].name, result.name)
+
     course_name = result.courses[0].name  # FE: 'Frans'
     id_ = result.identifier
 
     status = save("punten", course_name, id_, result, is_punten_json_the_same)
 
     if status == IsSaved.SAME:
+        logger.debug(" => Already processed")
         return
 
     text, subject = build_text(result=result, is_update=status != IsSaved.NEW)