Add PE lessons to the generated calendar

This is based on @FEgor04's work in PR #2.

- add PE lessons gathering and mapping
- refactor project structure to separate responsibilities
- switch to asyncio for io ops parallelization
- make logging more detailed
- improve type hints
- implement error handling

pyproject.toml

@@ -8,9 +8,10 @@ license = "MIT"
 [tool.poetry.dependencies]
 python = "^3.11"
 ics = "^0.7.2"
-Flask = "^2.0.2"
-requests = "^2.26.0"
+Flask = {extras = ["async"], version = "^2.3.3"}
 sentry-sdk = {extras = ["flask"], version = "^1.4.3"}
+aiohttp = "^3.8.5"
+async-cache = "^1.1.1"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.0.291"
@@ -36,7 +37,7 @@ files = ["src"]
 [[tool.mypy.overrides]]
 # https://mypy.readthedocs.io/en/stable/running_mypy.html#missing-imports
 # it's better to disable untyped modules on a per-module basis
-module = ["ics.*"]
+module = ["ics.*", "cache.*"]
 ignore_missing_imports = true
 
 

src/app.py

@@ -1,13 +1,17 @@
 import logging
+from asyncio import gather
+from itertools import chain
 
 import sentry_sdk
+from aiohttp import ClientSession
 from flask import Flask, Response
 from sentry_sdk.integrations.flask import FlaskIntegration
 
 from auth import get_access_token
-from calendar_processing import raw_events_to_calendar
+from calendar_processing import build_calendar, calendar_to_ics_text
 from credentials_hashing import get_credentials_hash
-from main_api import get_raw_events
+from lessons_to_events import raw_lesson_to_event, raw_pe_lesson_to_event
+from main_api import get_raw_lessons, get_raw_pe_lessons
 
 logging.basicConfig(level=logging.INFO)
 logging.getLogger("werkzeug").handlers = []  # prevent duplicated logging output
@@ -33,11 +37,25 @@
 
 
 @app.route(_calendar_route)
-def get_calendar():
-    token = get_access_token(app.config["ISU_USERNAME"], app.config["ISU_PASSWORD"])
-    events = get_raw_events(token)
-    calendar = raw_events_to_calendar(events)
-    return Response("\n".join(map(str.strip, calendar)), content_type="text/calendar")
+async def get_calendar():
+    async with ClientSession() as session:
+        token = await get_access_token(session, app.config["ISU_USERNAME"], app.config["ISU_PASSWORD"])
+
+        app.logger.info("Gathering lessons...")
+        lessons, pe_lessons = await gather(
+            get_raw_lessons(session, token),
+            get_raw_pe_lessons(session, token),
+        )
+        app.logger.info("Converting lessons to calendar events...")
+        lesson_events = map(raw_lesson_to_event, lessons)
+        pe_lesson_events = map(raw_pe_lesson_to_event, pe_lessons)
+
+        app.logger.info("Building calendar...")
+        calendar = build_calendar(chain(lesson_events, pe_lesson_events))
+        calendar_text = calendar_to_ics_text(calendar)
+
+        app.logger.info("Success, responding with calendar text...")
+        return Response(calendar_text, content_type="text/calendar")
 
 
 sentry_sdk.capture_message(f"my-itmo-ru-to-ical started for {app.config['ISU_USERNAME']}, hash {_creds_hash}")

src/auth.py

@@ -1,13 +1,15 @@
 import html
+import logging
 import os
 import re
 import urllib.parse
 from base64 import urlsafe_b64encode
 from hashlib import sha256
 
-import requests
+from aiohttp import ClientSession
+from cache import AsyncTTL
 
-from utils.timed_cache import timed_cache
+logger = logging.getLogger(__name__)
 
 # inspired by https://www.stefaanlippens.net/oauth-code-flow-pkce.html
 
@@ -30,12 +32,14 @@ def get_code_challenge(code_verifier: str):
 _FORM_ACTION_REGEX = re.compile(r'<form\s+.*?\s+action="(?P<action>.*?)"', re.DOTALL)
 
 
-@timed_cache(minutes=55)
-def get_access_token(username: str, password: str):
+@AsyncTTL(time_to_live=55 * 60, skip_args=1)
+async def get_access_token(session: ClientSession, username: str, password: str) -> str:
+    logger.info(f"Getting new access token for {username}")
+
     code_verifier = generate_code_verifier()
     code_challenge = get_code_challenge(code_verifier)
 
-    auth_resp = requests.get(
+    auth_resp = await session.get(
         _PROVIDER + "/protocol/openid-connect/auth",
         params=dict(
             protocol="oauth2",
@@ -50,25 +54,25 @@ def get_access_token(username: str, password: str):
     )
     auth_resp.raise_for_status()
 
-    form_action_match = _FORM_ACTION_REGEX.search(auth_resp.text)
+    form_action_match = _FORM_ACTION_REGEX.search(await auth_resp.text())
     assert form_action_match, "Keycloak form action regexp match not found"
     form_action = html.unescape(form_action_match.group("action"))
 
-    form_resp = requests.post(
+    form_resp = await session.post(
         url=form_action,
         data=dict(username=username, password=password),
         cookies=auth_resp.cookies,
         allow_redirects=False,
     )
-    if form_resp.status_code != 302:
-        raise ValueError(f"Wrong Keycloak form response: {form_resp.status_code} {form_resp.text}")
+    if form_resp.status != 302:
+        raise ValueError(f"Wrong Keycloak form response: {form_resp.status} {await form_resp.text()}")
 
     url_redirected_to = form_resp.headers["Location"]
     query = urllib.parse.urlparse(url_redirected_to).query
     redirect_params = urllib.parse.parse_qs(query)
     auth_code = redirect_params["code"][0]
 
-    token_resp = requests.post(
+    token_resp = await session.post(
         url=_PROVIDER + "/protocol/openid-connect/token",
         data=dict(
             grant_type="authorization_code",
@@ -80,5 +84,7 @@ def get_access_token(username: str, password: str):
         allow_redirects=False,
     )
     token_resp.raise_for_status()
-    result = token_resp.json()
-    return result["access_token"]
+    result = await token_resp.json()
+    token = result["access_token"]
+    logger.info(f"Got new access token for {username} successfully")
+    return token

src/calendar_processing.py

@@ -1,69 +1,21 @@
 from __future__ import annotations
 
-from datetime import datetime, timedelta
+import logging
+from typing import Iterable
 
-from dateutil.parser import isoparse
 from ics import Calendar, Event
 
-_event_type_to_tag_map = {
-    "Лекции": "Лек",
-    "Практические занятия": "Прак",
-    "Зачет": "Зачет",
-}
+logger = logging.getLogger(__name__)
 
-_raw_event_key_names = {
-    "group": "Группа",
-    "teacher_name": "Преподаватель",
-    "zoom_url": "Ссылка на Zoom",
-    "zoom_password": "Пароль Zoom",
-    "zoom_info": "Доп. информация для Zoom",
-    "note": "Примечание",
-}
 
-
-def _event_type_to_tag(t: str):
-    return _event_type_to_tag_map.get(t, t)
-
-
-def _raw_event_to_description(re: dict):
-    lines = []
-    for key, name in _raw_event_key_names.items():
-        if re[key]:
-            lines.append(f"{name}: {re[key]}")
-
-    _msk_formatted_datetime = (datetime.utcnow() + timedelta(hours=3)).strftime("%Y-%m-%d %H:%M")
-    lines.append(f"Обновлено: {_msk_formatted_datetime} MSK")
-    return "\n".join(lines)
-
-
-def _raw_event_to_location(re: dict):
-    elements = []
-    for key in "room", "building":
-        if re[key]:
-            elements.append(re[key])
-
-    result = ", ".join(elements)
-
-    if re["zoom_url"]:
-        result = f"Zoom / {result}" if result else "Zoom"
-
-    return result if result else None
-
-
-def raw_events_to_calendar(raw_events: list[dict]):
+def build_calendar(events: Iterable[Event]):
     calendar = Calendar()
     calendar.creator = "my-itmo-ru-to-ical"
-    for raw_event in raw_events:
-        event = Event(
-            name=f"[{_event_type_to_tag(raw_event['type'])}] {raw_event['subject']}",
-            begin=isoparse(f"{raw_event['date']}T{raw_event['time_start']}:00+03:00"),
-            end=isoparse(f"{raw_event['date']}T{raw_event['time_end']}:00+03:00"),
-            description=_raw_event_to_description(raw_event),
-            location=_raw_event_to_location(raw_event),
-        )
-        if raw_event["zoom_url"]:
-            event.url = raw_event["zoom_url"]
-
+    for event in events:
         calendar.events.add(event)
-
+    logger.info(f"Built a calendar with {len(calendar.events)} events")
     return calendar
+
+
+def calendar_to_ics_text(calendar: Calendar) -> str:
+    return "\n".join(map(str.strip, calendar))

src/lessons_to_events.py

@@ -0,0 +1,73 @@
+from datetime import datetime, timedelta
+
+from dateutil.parser import isoparse
+from ics import Event
+
+_lesson_type_to_tag_map = {
+    "Лекции": "Лек",
+    "Практические занятия": "Прак",
+    "Зачет": "Зачет",
+}
+
+_raw_lesson_key_names = {
+    "group": "Группа",
+    "teacher_name": "Преподаватель",
+    "teacher_fio": "Преподаватель",
+    "zoom_url": "Ссылка на Zoom",
+    "zoom_password": "Пароль Zoom",
+    "zoom_info": "Доп. информация для Zoom",
+    "note": "Примечание",
+}
+
+
+def _lesson_type_to_tag(t: str):
+    return _lesson_type_to_tag_map.get(t, t)
+
+
+def _raw_lesson_to_description(raw_lesson: dict):
+    lines = []
+    for key, name in _raw_lesson_key_names.items():
+        if raw_lesson.get(key):
+            lines.append(f"{name}: {raw_lesson[key]}")
+
+    _msk_formatted_datetime = (datetime.utcnow() + timedelta(hours=3)).strftime("%Y-%m-%d %H:%M")
+    lines.append(f"Обновлено: {_msk_formatted_datetime} MSK")
+    return "\n".join(lines)
+
+
+def _raw_lesson_to_location(raw_lesson: dict):
+    elements = []
+    for key in "room", "building":
+        if raw_lesson.get(key):
+            elements.append(raw_lesson[key])
+
+    result = ", ".join(elements)
+
+    if raw_lesson.get("zoom_url"):
+        result = f"Zoom / {result}" if result else "Zoom"
+
+    return result if result else None
+
+
+def raw_lesson_to_event(raw_lesson: dict) -> Event:
+    event = Event(
+        name=f"[{_lesson_type_to_tag(raw_lesson['type'])}] {raw_lesson['subject']}",
+        begin=isoparse(f"{raw_lesson['date']}T{raw_lesson['time_start']}:00+03:00"),
+        end=isoparse(f"{raw_lesson['date']}T{raw_lesson['time_end']}:00+03:00"),
+        description=_raw_lesson_to_description(raw_lesson),
+        location=_raw_lesson_to_location(raw_lesson),
+    )
+    if raw_lesson["zoom_url"]:
+        event.url = raw_lesson["zoom_url"]
+
+    return event
+
+
+def raw_pe_lesson_to_event(raw_pe_lesson: dict) -> Event:
+    return Event(
+        name=f"[Физра] {raw_pe_lesson['section_name']}",
+        begin=isoparse(f"{raw_pe_lesson['date']}"),
+        end=isoparse(f"{raw_pe_lesson['date_end']}"),
+        description=_raw_lesson_to_description(raw_pe_lesson),
+        location=raw_pe_lesson["room_name"],
+    )

src/main_api.py

@@ -1,8 +1,16 @@
 from __future__ import annotations
 
+import logging
 from datetime import date
+from typing import Iterable
 
-import requests
+from aiohttp import ClientSession
+
+from utils.error_handling import catch_errors_to_empty_iter
+
+logger = logging.getLogger(__name__)
+
+_API_BASE_URL = "https://my.itmo.ru/api"
 
 
 def _get_date_range_params() -> dict:
@@ -16,18 +24,25 @@ def _get_date_range_params() -> dict:
     )
 
 
-def get_raw_events(auth_token: str) -> list[dict]:
-    resp = requests.get(
-        "https://my.itmo.ru/api/schedule/schedule/personal",
-        params=_get_date_range_params(),
-        headers={"Authorization": f"Bearer {auth_token}"},
-    )
+async def _get_calendar(session: ClientSession, auth_token: str, path: str) -> dict:
+    url = _API_BASE_URL + path
+    params = _get_date_range_params()
+    logger.info(f"Getting calendar from {url}, using params {params}")
+
+    resp = await session.get(url, params=params, headers={"Authorization": f"Bearer {auth_token}"})
     resp.raise_for_status()
+    return await resp.json()
+
+
+@catch_errors_to_empty_iter
+async def get_raw_lessons(session: ClientSession, auth_token: str) -> Iterable[dict]:
+    resp_json = await _get_calendar(session, auth_token, "/schedule/schedule/personal")
+    days = resp_json["data"]
+    return (dict(date=day["date"], **lesson) for day in days for lesson in day["lessons"])
 
-    days = resp.json()["data"]
-    raw_events = []
-    for day in days:
-        for lesson in day["lessons"]:
-            raw_events.append(dict(date=day["date"], **lesson))
 
-    return raw_events
+@catch_errors_to_empty_iter
+async def get_raw_pe_lessons(session: ClientSession, auth_token: str) -> Iterable[dict]:
+    resp_json = await _get_calendar(session, auth_token, "/sport/my_sport/calendar")
+    days = resp_json["result"]
+    return (dict(**lesson) for day in days for lesson in day["lessons"])

src/utils/error_handling.py

@@ -0,0 +1,17 @@
+import logging
+from typing import Awaitable, Callable, Iterable, ParamSpec
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+
+
+def catch_errors_to_empty_iter(func: Callable[P, Awaitable[Iterable]]) -> Callable[P, Awaitable[Iterable]]:
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterable:
+        try:
+            return await func(*args, **kwargs)
+        except Exception:
+            logger.exception(f"Failed to call {func.__name__}")
+            return ()
+
+    return wrapper