Create app.py

app.py

@@ -0,0 +1,389 @@
+"""
+API PCN + PRONOTE — HuggingFace Spaces Free Tier
+Endpoints REST pour scraper ENT Paris Classe Numérique et PRONOTE.
+"""
+
+from __future__ import annotations
+
+import os
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+from enum import Enum
+
+from fastapi import FastAPI, HTTPException, Query, Depends
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+from pcn import Config as PCNConfig, ENTClient
+from pronote_client import Config as PronoteConfig, PronoteClient
+
+logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s")
+log = logging.getLogger("api")
+
+app = FastAPI(
+    title="PCN + PRONOTE API",
+    description="API REST pour récupérer notifications, messages, notes, devoirs, EDT depuis ENT PCN et PRONOTE.",
+    version="1.0.0",
+    docs_url="/",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  Modèles Pydantic
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+class Credentials(BaseModel):
+    login: str = Field(..., description="Identifiant")
+    password: str = Field(..., description="Mot de passe")
+
+
+class PCNRequest(Credentials):
+    hours: int = Field(24, ge=1, le=720, description="Fenêtre temporelle en heures")
+    fetch_body: bool = Field(True, description="Récupérer le corps des messages")
+    fetch_attachments: bool = Field(False, description="Télécharger les pièces jointes")
+
+
+class PronoteRequest(Credentials):
+    pronote_url: str = Field(..., description="URL PRONOTE élève")
+    ent: str = Field("", description="ENT (ex: ent_parisclassenumerique, vide=direct)")
+    hours: int = Field(168, ge=1, le=2160, description="Fenêtre temporelle en heures")
+    fetch_attachments: bool = Field(False, description="Télécharger les pièces jointes")
+
+
+class PronoteModules(str, Enum):
+    grades = "grades"
+    homework = "homework"
+    timetable = "timetable"
+    messages = "messages"
+    absences = "absences"
+    info = "info"
+
+
+class NotifFilter(str, Enum):
+    MESSAGERIE = "MESSAGERIE"
+    BLOG = "BLOG"
+    ACTUALITES = "ACTUALITES"
+    EXERCIZER = "EXERCIZER"
+    COMMUNITIES = "COMMUNITIES"
+    WIKI = "WIKI"
+    SCRAPBOOK = "SCRAPBOOK"
+    TIMELINEGENERATOR = "TIMELINEGENERATOR"
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  Helpers
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+def _pcn_client(req: PCNRequest) -> ENTClient:
+    cfg = PCNConfig()
+    cfg.login = req.login
+    cfg.password = req.password
+    cfg.hours_back = req.hours
+    cfg.fetch_body = req.fetch_body
+    cfg.fetch_attachments = req.fetch_attachments
+    client = ENTClient(cfg)
+    try:
+        client.login()
+    except SystemExit:
+        raise HTTPException(401, "Échec connexion PCN — identifiants invalides")
+    except Exception as e:
+        raise HTTPException(502, f"Erreur connexion PCN : {e}")
+    return client
+
+
+def _pronote_client(req: PronoteRequest) -> PronoteClient:
+    cfg = PronoteConfig()
+    cfg.pronote_url = req.pronote_url
+    cfg.login = req.login
+    cfg.password = req.password
+    cfg.ent = req.ent
+    cfg.hours_back = req.hours
+    cfg.fetch_attachments = req.fetch_attachments
+    cfg.dry_run = not req.fetch_attachments
+    client = PronoteClient(cfg)
+    try:
+        client.login()
+    except SystemExit:
+        raise HTTPException(401, "Échec connexion PRONOTE — identifiants ou URL invalides")
+    except Exception as e:
+        raise HTTPException(502, f"Erreur connexion PRONOTE : {e}")
+    return client
+
+
+def _serialize(obj):
+    """Convertit dataclasses en dicts récursivement."""
+    from dataclasses import asdict, is_dataclass
+    if is_dataclass(obj):
+        return asdict(obj)
+    if isinstance(obj, list):
+        return [_serialize(x) for x in obj]
+    return obj
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  Health
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+@app.get("/health", tags=["Système"])
+def health():
+    return {"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()}
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  PCN — Endpoints
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+@app.post("/pcn/all", tags=["PCN"], summary="Tout récupérer (notifications + messages)")
+def pcn_all(req: PCNRequest):
+    client = _pcn_client(req)
+    try:
+        notifs = client.fetch_notifications()
+        raw = client.fetch_messages()
+        msgs = client.process(raw)
+        report = client.build_report(notifs, msgs)
+        return _serialize(report)
+    finally:
+        client.close()
+
+
+@app.post("/pcn/notifications", tags=["PCN"], summary="Notifications récentes")
+def pcn_notifications(
+    req: PCNRequest,
+    type_filter: Optional[list[NotifFilter]] = Query(None, alias="type", description="Filtrer par type"),
+    sender: Optional[str] = Query(None, description="Filtrer par expéditeur (contient)"),
+    limit: int = Query(100, ge=1, le=500, description="Nombre max de résultats"),
+):
+    client = _pcn_client(req)
+    try:
+        notifs = client.fetch_notifications()
+
+        if type_filter:
+            allowed = {t.value for t in type_filter}
+            notifs = [n for n in notifs if n.type in allowed]
+
+        if sender:
+            s = sender.lower()
+            notifs = [n for n in notifs if s in n.sender.lower()]
+
+        notifs = notifs[:limit]
+        return {
+            "count": len(notifs),
+            "notifications": _serialize(notifs),
+        }
+    finally:
+        client.close()
+
+
+@app.post("/pcn/messages", tags=["PCN"], summary="Messages non lus")
+def pcn_messages(
+    req: PCNRequest,
+    sender: Optional[str] = Query(None, description="Filtrer par expéditeur (contient)"),
+    subject: Optional[str] = Query(None, description="Filtrer par sujet (contient)"),
+    role: Optional[str] = Query(None, description="Filtrer par rôle (Teacher, Student, Relative…)"),
+    has_attachments: Optional[bool] = Query(None, description="Avec pièces jointes uniquement"),
+    limit: int = Query(50, ge=1, le=200),
+):
+    client = _pcn_client(req)
+    try:
+        raw = client.fetch_messages()
+        msgs = client.process(raw)
+
+        if sender:
+            s = sender.lower()
+            msgs = [m for m in msgs if s in m.sender.lower()]
+        if subject:
+            s = subject.lower()
+            msgs = [m for m in msgs if s in m.subject.lower()]
+        if role:
+            r = role.lower()
+            msgs = [m for m in msgs if r in m.role.lower()]
+        if has_attachments is not None:
+            msgs = [m for m in msgs if m.has_attachments == has_attachments]
+
+        msgs = msgs[:limit]
+        return {
+            "count": len(msgs),
+            "messages": _serialize(msgs),
+        }
+    finally:
+        client.close()
+
+
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  PRONOTE — Endpoints
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+@app.post("/pronote/all", tags=["PRONOTE"], summary="Tout récupérer")
+def pronote_all(
+    req: PronoteRequest,
+    modules: Optional[list[PronoteModules]] = Query(None, description="Modules à récupérer (défaut: tous)"),
+):
+    client = _pronote_client(req)
+    try:
+        mods = {m.value for m in modules} if modules else {"grades", "homework", "timetable", "messages", "absences", "info"}
+
+        grades = client.fetch_grades() if "grades" in mods else []
+        homework = client.fetch_homework() if "homework" in mods else []
+        timetable = client.fetch_timetable() if "timetable" in mods else []
+        messages = client.fetch_messages() if "messages" in mods else []
+        absences = client.fetch_absences() if "absences" in mods else []
+        info = client.fetch_info() if "info" in mods else []
+
+        report = client.build_report(grades, homework, timetable, messages, absences, info)
+        return _serialize(report)
+    finally:
+        client.close()
+
+
+@app.post("/pronote/grades", tags=["PRONOTE"], summary="Notes")
+def pronote_grades(
+    req: PronoteRequest,
+    subject: Optional[str] = Query(None, description="Filtrer par matière (contient)"),
+    min_grade: Optional[float] = Query(None, description="Note minimale"),
+    max_grade: Optional[float] = Query(None, description="Note maximale"),
+    limit: int = Query(100, ge=1, le=500),
+):
+    client = _pronote_client(req)
+    try:
+        grades = client.fetch_grades()
+
+        if subject:
+            s = subject.lower()
+            grades = [g for g in grades if s in g.subject.lower()]
+        if min_grade is not None:
+            grades = [g for g in grades if _parse_num(g.grade) >= min_grade]
+        if max_grade is not None:
+            grades = [g for g in grades if _parse_num(g.grade) <= max_grade]
+
+        grades = grades[:limit]
+        return {"count": len(grades), "grades": _serialize(grades)}
+    finally:
+        client.close()
+
+
+@app.post("/pronote/homework", tags=["PRONOTE"], summary="Devoirs")
+def pronote_homework(
+    req: PronoteRequest,
+    subject: Optional[str] = Query(None, description="Filtrer par matière"),
+    done: Optional[bool] = Query(None, description="Filtrer fait/non fait"),
+    limit: int = Query(100, ge=1, le=500),
+):
+    client = _pronote_client(req)
+    try:
+        hw = client.fetch_homework()
+
+        if subject:
+            s = subject.lower()
+            hw = [h for h in hw if s in h.subject.lower()]
+        if done is not None:
+            hw = [h for h in hw if h.done == done]
+
+        hw = hw[:limit]
+        return {"count": len(hw), "homework": _serialize(hw)}
+    finally:
+        client.close()
+
+
+@app.post("/pronote/timetable", tags=["PRONOTE"], summary="Emploi du temps (14 jours)")
+def pronote_timetable(
+    req: PronoteRequest,
+    subject: Optional[str] = Query(None, description="Filtrer par matière"),
+    teacher: Optional[str] = Query(None, description="Filtrer par professeur"),
+    cancelled: Optional[bool] = Query(None, description="Cours annulés uniquement"),
+    date: Optional[str] = Query(None, description="Filtrer par date (YYYY-MM-DD)"),
+):
+    client = _pronote_client(req)
+    try:
+        lessons = client.fetch_timetable()
+
+        if subject:
+            s = subject.lower()
+            lessons = [l for l in lessons if s in l.subject.lower()]
+        if teacher:
+            t = teacher.lower()
+            lessons = [l for l in lessons if t in l.teacher.lower()]
+        if cancelled is not None:
+            lessons = [l for l in lessons if l.is_cancelled == cancelled]
+        if date:
+            lessons = [l for l in lessons if l.start.startswith(date)]
+
+        return {"count": len(lessons), "timetable": _serialize(lessons)}
+    finally:
+        client.close()
+
+
+@app.post("/pronote/messages", tags=["PRONOTE"], summary="Messages")
+def pronote_messages(
+    req: PronoteRequest,
+    sender: Optional[str] = Query(None, description="Filtrer par expéditeur"),
+    unread: Optional[bool] = Query(None, description="Non lus uniquement"),
+    limit: int = Query(50, ge=1, le=200),
+):
+    client = _pronote_client(req)
+    try:
+        msgs = client.fetch_messages()
+
+        if sender:
+            s = sender.lower()
+            msgs = [m for m in msgs if s in m.sender.lower()]
+        if unread is not None:
+            msgs = [m for m in msgs if (not m.read) == unread]
+
+        msgs = msgs[:limit]
+        return {"count": len(msgs), "messages": _serialize(msgs)}
+    finally:
+        client.close()
+
+
+@app.post("/pronote/absences", tags=["PRONOTE"], summary="Absences")
+def pronote_absences(
+    req: PronoteRequest,
+    justified: Optional[bool] = Query(None, description="Justifiée ou non"),
+):
+    client = _pronote_client(req)
+    try:
+        absences = client.fetch_absences()
+
+        if justified is not None:
+            absences = [a for a in absences if a.justified == justified]
+
+        return {"count": len(absences), "absences": _serialize(absences)}
+    finally:
+        client.close()
+
+
+@app.post("/pronote/info", tags=["PRONOTE"], summary="Informations scolaires")
+def pronote_info(
+    req: PronoteRequest,
+    limit: int = Query(50, ge=1, le=200),
+):
+    client = _pronote_client(req)
+    try:
+        info = client.fetch_info()
+        info = info[:limit]
+        return {"count": len(info), "info": _serialize(info)}
+    finally:
+        client.close(
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+#  Utils
+# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+
+def _parse_num(val: str) -> float:
+    try:
+        return float(val.replace(",", "."))
+    except (ValueError, AttributeError):
+        return 0.0