Big update. Add LMS integration, views. Edit bulk schema

backend/yet_another_calendar/web/api/bulk/integration.py

@@ -44,12 +44,13 @@ def export_to_ics(calendar: schema.CalendarResponse) -> Iterable[bytes]:
     for netology_lesson in calendar.netology.webinars:
         if not netology_lesson.starts_at or not netology_lesson.ends_at:
             continue
-        event = create_ics_event(title=f"Netology: {netology_lesson.title}", starts_at=netology_lesson.starts_at,
+        event = create_ics_event(title=f"Netology: {netology_lesson.block_title}|{netology_lesson.title}",
+                                 starts_at=netology_lesson.starts_at,
                                  ends_at=netology_lesson.ends_at, lesson_id=netology_lesson.id,
                                  webinar_url=netology_lesson.webinar_url)
         ics_calendar.add_component(event)
-    for modeus_lesson in calendar.modeus:
-        event = create_ics_event(title=f"Modeus: {modeus_lesson.name}",
+    for modeus_lesson in calendar.utmn.modeus_events:
+        event = create_ics_event(title=f"Modeus: {modeus_lesson.course_name}|{modeus_lesson.name}",
                                  starts_at=modeus_lesson.start_time, ends_at=modeus_lesson.end_time,
                                  lesson_id=modeus_lesson.id,
                                  description=modeus_lesson.description)
@@ -101,7 +102,9 @@ async def get_calendar(
         netology_response = tg.create_task(netology_views.get_calendar(body, calendar_id, cookies))
         modeus_response = tg.create_task(modeus_views.get_calendar(body, jwt_token))
     return schema.CalendarResponse.model_validate(
-        {"netology": netology_response.result(), "modeus": modeus_response.result()},
+        {"netology": netology_response.result(), "utmn": {
+            "modeus_events": modeus_response.result(),
+        }},
     ).change_timezone(tz)
 
 

backend/yet_another_calendar/web/api/bulk/schema.py

@@ -1,16 +1,24 @@
 import datetime
 import hashlib
-from typing import Self
+from typing import Self, Optional
 
 from pydantic import BaseModel, Field
 
 from ..modeus import schema as modeus_schema
+from ..lms import schema as lms_schema
 from ..netology import schema as netology_schema
 
+def now_dt_utc() -> datetime.datetime:
+    return datetime.datetime.now(tz=datetime.timezone.utc)
+
+class UtmnResponse(BaseModel):
+    modeus_events: list[modeus_schema.FullEvent]
+    lms_events: Optional[list[lms_schema.ModuleResponse]] = Field(default=None)
+
 
 class BulkResponse(BaseModel):
     netology: netology_schema.SerializedEvents
-    modeus: list[modeus_schema.FullEvent]
+    utmn: UtmnResponse
 
     def change_timezone(self, timezone: datetime.tzinfo) -> Self:
         for homework in self.netology.homework:
@@ -20,13 +28,14 @@ def change_timezone(self, timezone: datetime.tzinfo) -> Self:
             webinar.starts_at = webinar.validate_starts_at(webinar.starts_at, timezone)
             webinar.ends_at = webinar.validate_ends_at(webinar.ends_at, timezone)
 
-        for event in self.modeus:
+        for event in self.utmn.modeus_events:
             event.start_time = event.validate_starts_at(event.start_time, timezone)
             event.end_time = event.validate_end_time(event.end_time, timezone)
         return self
 
+
 class CalendarResponse(BulkResponse):
-    cached_at: datetime.datetime = Field(default_factory=datetime.datetime.now, alias="cached_at")
+    cached_at: datetime.datetime = Field(default_factory=now_dt_utc, alias="cached_at")
 
     def get_hash(self) -> str:
         dump = BulkResponse(**self.model_dump(by_alias=True)).model_dump_json(by_alias=True)

backend/yet_another_calendar/web/api/lms/__init__.py

@@ -0,0 +1,3 @@
+from .views import router
+
+__all__ = ["router"]

backend/yet_another_calendar/web/api/lms/integration.py

@@ -0,0 +1,127 @@
+"""Netology API implementation."""
+import asyncio
+from typing import Any
+
+import httpx
+import reretry
+from fastapi import HTTPException
+from httpx import AsyncClient
+from pydantic import TypeAdapter
+from starlette import status
+
+from . import schema
+from yet_another_calendar.settings import settings
+from ..modeus.schema import ModeusTimeBody
+
+
+@reretry.retry(exceptions=httpx.TransportError, tries=settings.retry_tries, delay=settings.retry_delay)
+async def get_token(creds: schema.LxpCreds, timeout: int = 15) -> str:
+    """
+    Auth in lms, required username and password.
+    """
+    session = AsyncClient(
+        http2=True,
+        base_url="https://lms.utmn.ru",
+        timeout=timeout,
+    )
+    response = await session.post('/login/token.php', data=creds.model_dump())
+    response.raise_for_status()
+    serialized_response = response.json()
+    error = serialized_response.get('error') or serialized_response.get('exception')
+    if error:
+        raise HTTPException(detail=f'{error}. Server response: {serialized_response}',
+                            status_code=response.status_code)
+    return serialized_response['token']
+
+
+@reretry.retry(exceptions=httpx.TransportError, tries=settings.retry_tries, delay=settings.retry_delay)
+async def send_request(
+        request_settings: dict[str, Any], timeout: int = 15) -> dict[str, Any] | list[dict[str, Any]]:
+    """Send request from httpx."""
+    session = AsyncClient(
+        http2=True,
+        base_url="https://lms.utmn.ru",
+        timeout=timeout,
+    )
+    response = await session.request(**request_settings)
+    response.raise_for_status()
+    serialized_response = response.json()
+    if isinstance(serialized_response, list):
+        return serialized_response
+    error = serialized_response.get('error') or serialized_response.get('exception')
+    if error:
+        raise HTTPException(detail=f'{error}. Server response: {serialized_response}',
+                            status_code=status.HTTP_400_BAD_REQUEST)
+    return serialized_response
+
+
+async def get_user_info(token: str, username: str) -> list[dict[str, Any]]:
+    response = await send_request(
+        request_settings={
+            'method': 'POST',
+            'url': '/webservice/rest/server.php',
+            'params': {"wsfunction": "core_user_get_users_by_field",
+                       "field": "username",
+                       "values[0]": username,
+                       "wstoken": token,
+                       "moodlewsrestformat": "json"},
+        })
+    return response
+
+
+async def auth_lms(creds: schema.LxpCreds) -> schema.User:
+    """Get token and username"""
+    token = await get_token(creds)
+    user_info = await get_user_info(token, creds.get_username())
+    return schema.User(**user_info[0], token=token)
+
+
+async def get_courses(user: schema.User) -> list[schema.Course]:
+    """Get courses."""
+    response = await send_request(
+        request_settings={
+            'method': 'GET',
+            'url': '/webservice/rest/server.php',
+            'params': {
+                'wstoken': user.token,
+                'moodlewsrestformat': 'json',
+                'wsfunction': 'core_enrol_get_users_courses',
+                'userid': user.id,
+            },
+        })
+    adapter = TypeAdapter(list[schema.Course])
+    return adapter.validate_python(response)
+
+
+async def get_extended_course(user: schema.User, course_id: int) -> list[schema.ExtendedCourse]:
+    """Get extended course with modules and deadlines."""
+    response = await send_request(
+        request_settings={
+            'method': 'POST',
+            'url': '/webservice/rest/server.php',
+            'params': {
+                'wstoken': user.token,
+                'wsfunction': 'core_course_get_contents',
+                'courseid': course_id,
+                'moodlewsrestformat': 'json',
+            },
+        })
+    adapter = TypeAdapter(list[schema.ExtendedCourse])
+    return adapter.validate_python(response)
+
+
+async def get_filtered_courses(user: schema.User, body: ModeusTimeBody) -> list[schema.ModuleResponse]:
+    """Filter LXP events."""
+    courses = await get_courses(user)
+    course_by_ids = {course.id: course for course in courses}
+    tasks = {}
+    async with asyncio.TaskGroup() as tg:
+        for course in courses:
+            tasks[course.id] = tg.create_task(get_extended_course(user, course.id))
+    filtered_modules = []
+    for course_id, task in tasks.items():
+        course_name = course_by_ids[course_id].full_name
+        extended_course = task.result()
+        for module in extended_course:
+            filtered_modules.extend(module.get_filtered_modules(body, course_name))
+    return filtered_modules

backend/yet_another_calendar/web/api/lms/schema.py

@@ -0,0 +1,106 @@
+import datetime
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field, model_validator
+
+from yet_another_calendar.web.api.modeus.schema import Creds, ModeusTimeBody
+
+
+class LxpCreds(Creds):
+    service: str = "test"
+
+    def get_username(self) -> str:
+        return self.username.split("@")[0]
+
+
+class User(BaseModel):
+    id: int
+    token: str
+
+
+class Course(BaseModel):
+    id: int
+    short_name: str = Field(alias="shortname")
+    full_name: str = Field(alias="fullname")
+    completed: bool
+    hidden: bool
+
+
+class ModuleState(BaseModel):
+    state: bool
+
+
+class DateModule(BaseModel):
+    label: str
+    date: datetime.datetime = Field(alias="timestamp")
+    dataid: str
+
+    @model_validator(mode='before')
+    @classmethod
+    def deadline_validation(cls, data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+        timestamp = data.get('timestamp')
+        if timestamp is None:
+            return data
+        data['timestamp'] = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
+        return data
+
+
+class BaseModule(BaseModel):
+    id: int
+    url: Optional[str] = Field(default=None)
+    name: str
+    user_visible: bool = Field(alias="uservisible")
+    modname: str
+
+
+class Module(BaseModule):
+    dates: list[DateModule]
+    completion_state: Optional[ModuleState] = Field(alias="completiondata", default=None)
+
+
+class ModuleResponse(BaseModule):
+    dt_start: Optional[datetime.datetime]
+    dt_end: Optional[datetime.datetime]
+    is_completed: bool
+    course_name: str
+
+
+class ExtendedCourse(BaseModel):
+    id: int
+    name: str
+    modules: list[Module]
+
+    @staticmethod
+    def is_suitable_time(deadline: datetime.datetime,
+                         time_min: datetime.datetime, time_max: datetime.datetime) -> bool:
+        """Check if lesson have suitable time"""
+        if deadline and time_max > deadline > time_min:
+            return True
+        return False
+
+    def get_filtered_modules(self, body: ModeusTimeBody, course_name: str) -> list[ModuleResponse]:
+        """Filter module by time and user_visible."""
+        filtered_modules = []
+        for module in self.modules:
+            dt_end = None
+            dt_start = None
+            if module.dates and len(module.dates) > 1:
+                dt_start = module.dates[0].date
+                dt_end = module.dates[1].date
+            else:
+                continue
+            if self.is_suitable_time(dt_end, body.time_min, body.time_max) and module.user_visible:
+                completion_state = module.completion_state
+                state = False
+                if completion_state:
+                    state = completion_state.state
+                filtered_modules.append(ModuleResponse(
+                    **module.model_dump(by_alias=True),
+                    is_completed=state,
+                    dt_end=dt_end, dt_start=dt_start,
+                    course_name=course_name,
+
+                ))
+        return filtered_modules

backend/yet_another_calendar/web/api/lms/views.py

@@ -0,0 +1,48 @@
+from fastapi import APIRouter
+
+from . import integration, schema
+from ..modeus.schema import ModeusTimeBody
+
+router = APIRouter()
+
+
+@router.post("/auth")
+async def get_netology_cookies(
+        creds: schema.LxpCreds,
+) -> schema.User:
+    """
+    Auth in LXP and return token.
+    """
+    return await integration.auth_lms(creds)
+
+
+@router.post("/courses")
+async def get_courses(
+        user: schema.User,
+) -> list[schema.Course]:
+    """
+    Get LMS courses for current user.
+    """
+    return await integration.get_courses(user)
+
+
+@router.post("/course_info")
+async def get_user_info(
+        user: schema.User,
+        course_id: int = 2745,
+) -> list[schema.ExtendedCourse]:
+    """
+    Get LMS course info for current user.
+    """
+    return await integration.get_extended_course(user, course_id)
+
+
+@router.post("/events")
+async def get_course_info(
+        user: schema.User,
+        body: ModeusTimeBody,
+) -> list[schema.ModuleResponse]:
+    """
+    Get LMS events for current user.
+    """
+    return await integration.get_filtered_courses(user, body)