chore(integration): intégrer les PR Pronote critiques dans main
Revue Codex validée. Intégration dans main de la chaîne des PR #31, #32 et #33. Co-authored-by: Codex <codex@antoineve.me> Co-committed-by: Codex <codex@antoineve.me>
This commit was merged in pull request #35.
This commit is contained in:
@@ -2,16 +2,19 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, SchoolEvent
|
||||
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message
|
||||
from pronote_sync.sources.pronote.fallback import PronoteFetcherProtocol
|
||||
from pronote_sync.utils.redaction import redact_exception
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FetchedPronoteData:
|
||||
@@ -36,27 +39,41 @@ def resolve_target_date(
|
||||
) -> date:
|
||||
"""Détermine la date cible du digest à partir de l'agenda disponible.
|
||||
|
||||
La règle privilégie J+1 lorsqu'il contient des cours. Si la journée en
|
||||
cours contient des cours mais pas J+1, le prochain cours connu est choisi.
|
||||
Sans cours correspondant, J+1 est conservé, y compris pendant les vacances.
|
||||
La règle privilégie J+1 lorsqu'il contient un cours non annulé. Sinon,
|
||||
elle choisit le prochain jour possédant un cours effectif dans la fenêtre.
|
||||
Si aucune reprise n'est visible, la fin exclusive d'un événement scolaire
|
||||
couvrant J+1 est utilisée ; à défaut, J+1 reste le meilleur choix connu.
|
||||
|
||||
:param today: Date de référence de l'exécution.
|
||||
:param lessons: Cours récupérés pour la fenêtre de synchronisation.
|
||||
:param school_events: Événements scolaires récupérés (réservés aux évolutions
|
||||
du libellé de jour sans cours).
|
||||
:param school_events: Événements scolaires récupérés, avec borne de fin exclusive.
|
||||
:return: Date cible du digest.
|
||||
:rtype: date
|
||||
"""
|
||||
del school_events
|
||||
tomorrow = date.fromordinal(today.toordinal() + 1)
|
||||
lesson_dates = {lesson.start.date() for lesson in lessons}
|
||||
if tomorrow in lesson_dates:
|
||||
return tomorrow
|
||||
if today in lesson_dates:
|
||||
future_dates = sorted(day for day in lesson_dates if day > today)
|
||||
if future_dates:
|
||||
return future_dates[0]
|
||||
return tomorrow
|
||||
effective_dates = {
|
||||
lesson.start.date()
|
||||
for lesson in lessons
|
||||
if lesson.status != LessonStatus.CANCELLED and lesson.start.date() > today
|
||||
}
|
||||
if tomorrow in effective_dates:
|
||||
reason = "cours effectif demain"
|
||||
target = tomorrow
|
||||
elif effective_dates:
|
||||
reason = "prochain cours effectif connu"
|
||||
target = min(effective_dates)
|
||||
else:
|
||||
event_ends = [
|
||||
event.to_date for event in school_events if event.from_date <= tomorrow < event.to_date
|
||||
]
|
||||
if event_ends:
|
||||
reason = "fin exclusive de l'événement couvrant demain"
|
||||
target = max(event_ends)
|
||||
else:
|
||||
reason = "aucune reprise connue"
|
||||
target = tomorrow
|
||||
logger.info("Date cible %s : %s", target.isoformat(), reason)
|
||||
return target
|
||||
|
||||
|
||||
def _fetch_optional_messages(
|
||||
|
||||
@@ -25,10 +25,11 @@ import requests
|
||||
|
||||
from pronote_sync.config.settings import PronoteSettings
|
||||
from pronote_sync.errors import PronoteAuthRotationError
|
||||
from pronote_sync.models.agenda import Lesson, LessonStatus
|
||||
from pronote_sync.models.agenda import Lesson
|
||||
from pronote_sync.models.homework import Homework
|
||||
from pronote_sync.models.message import Message, MessageType
|
||||
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
|
||||
from pronote_sync.sources.pronote.lessons import collapse_replaced_lessons, lesson_status
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
|
||||
|
||||
@@ -573,12 +574,12 @@ class PronoteClient:
|
||||
teachers=tuple(lesson.teacher_names or ()),
|
||||
rooms=tuple(lesson.classrooms or ()),
|
||||
group=lesson.group_name,
|
||||
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
|
||||
status=lesson_status(getattr(lesson, "status", None), lesson.canceled),
|
||||
content=content.description if content is not None else None,
|
||||
)
|
||||
)
|
||||
self._persist_credentials()
|
||||
return lessons
|
||||
return collapse_replaced_lessons(lessons)
|
||||
|
||||
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||
"""Récupère les devoirs via ``pronotepy``.
|
||||
|
||||
@@ -35,6 +35,7 @@ from ...models.agenda import (
|
||||
from ...models.homework import Homework
|
||||
from ...utils.redaction import redact_exception, redact_url
|
||||
from ...utils.uid import generate_deterministic_uid, normalize_pronote_uid
|
||||
from .lessons import collapse_replaced_lessons
|
||||
|
||||
_HEADER_LABEL_PATTERN = re.compile(
|
||||
r"(?P<label>Mati(?:ère|ere)|Professeur(?:s|\(s\))?|Salle(?:s|\(s\))?"
|
||||
@@ -186,11 +187,12 @@ def parse_header(header: str) -> HeaderInfo:
|
||||
"group": None,
|
||||
"class_part": None,
|
||||
}
|
||||
matches = list(_HEADER_LABEL_PATTERN.finditer(header))
|
||||
header_text = _strip_html(header)
|
||||
matches = list(_HEADER_LABEL_PATTERN.finditer(header_text))
|
||||
for index, match in enumerate(matches):
|
||||
value_start = match.end()
|
||||
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(header)
|
||||
value = unescape(header[value_start:value_end].strip())
|
||||
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(header_text)
|
||||
value = unescape(header_text[value_start:value_end].strip())
|
||||
label = _normalize_label(match.group("label")).replace("(s)", "s")
|
||||
if label == "matiere":
|
||||
info["subject"] = value
|
||||
@@ -567,4 +569,4 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
|
||||
)
|
||||
)
|
||||
|
||||
return lessons, homeworks, school_events
|
||||
return collapse_replaced_lessons(lessons), homeworks, school_events
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Règles communes de statut et de consolidation des cours Pronote."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unicodedata
|
||||
from collections import defaultdict
|
||||
|
||||
from ...models.agenda import Lesson, LessonStatus
|
||||
|
||||
|
||||
def _normalize_status(value: str) -> str:
|
||||
"""Normalise un statut Pronote pour une comparaison tolérante.
|
||||
|
||||
:param value: Statut brut fourni par Pronote.
|
||||
:return: Statut minuscule sans accents.
|
||||
:rtype: str
|
||||
"""
|
||||
decomposed = unicodedata.normalize("NFKD", value)
|
||||
without_accents = "".join(char for char in decomposed if not unicodedata.combining(char))
|
||||
return " ".join(without_accents.casefold().split())
|
||||
|
||||
|
||||
def lesson_status(raw_status: object, canceled: bool = False) -> LessonStatus:
|
||||
"""Convertit le statut Pronote en statut métier canonique.
|
||||
|
||||
L'indicateur d'annulation et les libellés annulé sont prioritaires sur un
|
||||
éventuel libellé de déplacement. Les libellés inconnus restent normaux.
|
||||
|
||||
:param raw_status: Valeur du champ ``Lesson.status`` de pronotepy.
|
||||
:param canceled: Indicateur ``Lesson.canceled`` de pronotepy.
|
||||
:return: Statut métier canonique.
|
||||
:rtype: LessonStatus
|
||||
"""
|
||||
status = _normalize_status(raw_status) if isinstance(raw_status, str) else ""
|
||||
if canceled or "annul" in status:
|
||||
return LessonStatus.CANCELLED
|
||||
if any(token in status for token in ("deplac", "changement de salle", "modifi")):
|
||||
return LessonStatus.MOVED
|
||||
return LessonStatus.NORMAL
|
||||
|
||||
|
||||
def collapse_replaced_lessons(lessons: list[Lesson]) -> list[Lesson]:
|
||||
"""Supprime l'ancien exemplaire annulé lorsqu'un cours déplacé est présent.
|
||||
|
||||
Deux cours sont considérés comme les exemplaires d'une même modification
|
||||
lorsqu'ils partagent les dates, la matière, les enseignants et le groupe,
|
||||
et qu'une seule annulation et un seul déplacement se distinguent par leur
|
||||
salle. Une annulation isolée ou ambiguë est toujours conservée.
|
||||
|
||||
:param lessons: Cours issus d'une source Pronote.
|
||||
:return: Cours consolidés dans leur ordre d'origine.
|
||||
:rtype: list[Lesson]
|
||||
"""
|
||||
grouped: dict[tuple[object, ...], list[tuple[int, Lesson]]] = defaultdict(list)
|
||||
for index, lesson in enumerate(lessons):
|
||||
key = (
|
||||
lesson.start,
|
||||
lesson.end,
|
||||
lesson.subject.casefold(),
|
||||
tuple(teacher.casefold() for teacher in lesson.teachers),
|
||||
(lesson.group or "").casefold(),
|
||||
)
|
||||
grouped[key].append((index, lesson))
|
||||
|
||||
replaced_indexes: set[int] = set()
|
||||
for siblings in grouped.values():
|
||||
cancelled = [item for item in siblings if item[1].status == LessonStatus.CANCELLED]
|
||||
moved = [item for item in siblings if item[1].status == LessonStatus.MOVED]
|
||||
if len(cancelled) == 1 and len(moved) == 1 and cancelled[0][1].rooms != moved[0][1].rooms:
|
||||
replaced_indexes.add(cancelled[0][0])
|
||||
|
||||
return [lesson for index, lesson in enumerate(lessons) if index not in replaced_indexes]
|
||||
Reference in New Issue
Block a user