432 lines
15 KiB
Python
432 lines
15 KiB
Python
|
"""
|
||
|
|
Client PRONOTE pour GMAO Collège
|
||
|
|
Utilise la bibliothèque pronotepy pour se connecter à Pronote via QR Code
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import uuid
|
||
|
|
from datetime import datetime, timezone, timedelta
|
||
|
|
from typing import Optional, Dict, Any
|
||
|
|
|
||
|
|
# Import de pronotepy bundlée - la classe s'appelle 'Client'
|
||
|
|
# VieScolaireClient pour les établissements scolaires
|
||
|
|
from .pronotepy import Client as PronoteClient, VieScolaireClient
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
# Client global en mémoire
|
||
|
|
_pronote_client: Optional[PronoteClient] = None
|
||
|
|
_pronote_session: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
|
||
|
|
def _get_or_create_device_uuid() -> str:
|
||
|
|
"""
|
||
|
|
Récupère l'UUID stocké en base ou en génère un nouveau.
|
||
|
|
L'UUID doit rester constant pour que PRONOTE reconnaisse l'appareil.
|
||
|
|
Ne crée PAS de session partielle - juste retourne l'UUID.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from app_new.pronote.models import PronoteSession
|
||
|
|
|
||
|
|
session = PronoteSession.get()
|
||
|
|
if session and session.device_uuid:
|
||
|
|
logger.info(f"UUID existant récupéré: {session.device_uuid}")
|
||
|
|
return session.device_uuid
|
||
|
|
|
||
|
|
# Générer un nouvel UUID (sera sauvegardé lors de la connexion)
|
||
|
|
new_uuid = str(uuid.uuid4())
|
||
|
|
logger.info(f"Nouveau UUID généré: {new_uuid}")
|
||
|
|
return new_uuid
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur récupération UUID: {e}")
|
||
|
|
# Fallback: générer un UUID temporaire
|
||
|
|
return str(uuid.uuid4())
|
||
|
|
|
||
|
|
|
||
|
|
def connect_with_qr(qr_code_json: str, pin: str, account_pin: Optional[str] = None) -> Optional[PronoteClient]:
|
||
|
|
"""
|
||
|
|
Connecte à Pronote via un QR Code JSON.
|
||
|
|
|
||
|
|
IMPORTANT: Une connexion QR est considérée comme un "nouvel appareil" par PRONOTE.
|
||
|
|
On génère donc TOUJOURS un nouvel UUID pour chaque connexion QR.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
qr_code_json: JSON du QR code (string ou dict)
|
||
|
|
pin: PIN du QR Code (4 chiffres)
|
||
|
|
account_pin: PIN du compte si différent du PIN QR
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
PronoteClient si connexion réussie, None sinon
|
||
|
|
"""
|
||
|
|
global _pronote_client, _pronote_session
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Parser le JSON si c'est une string
|
||
|
|
if isinstance(qr_code_json, str):
|
||
|
|
qr_data = json.loads(qr_code_json)
|
||
|
|
else:
|
||
|
|
qr_data = qr_code_json
|
||
|
|
|
||
|
|
# Vérifier les clés requises
|
||
|
|
required_keys = ['login', 'jeton', 'url']
|
||
|
|
if not all(k in qr_data for k in required_keys):
|
||
|
|
raise ValueError("Le QR Code doit contenir les clés: login, jeton, url")
|
||
|
|
|
||
|
|
# IMPORTANT: Pour une connexion QR, PRONOTE attend un NOUVEL UUID
|
||
|
|
# (c'est un "nouvel appareil" pour lui)
|
||
|
|
device_uuid = str(uuid.uuid4())
|
||
|
|
logger.info(f"Nouvel UUID généré pour connexion QR: {device_uuid}")
|
||
|
|
|
||
|
|
# Utiliser le PIN du compte si fourni, sinon le PIN QR
|
||
|
|
effective_pin = account_pin if account_pin else pin
|
||
|
|
|
||
|
|
# Appliquer la méthode de classe qrcode_login
|
||
|
|
_pronote_client = PronoteClient.qrcode_login(
|
||
|
|
qr_code=qr_data,
|
||
|
|
pin=pin,
|
||
|
|
uuid=device_uuid,
|
||
|
|
account_pin=effective_pin,
|
||
|
|
device_name='GMAO College'
|
||
|
|
)
|
||
|
|
|
||
|
|
# Stocker les infos de session
|
||
|
|
# Utiliser client.password directement (comme l'ancien code)
|
||
|
|
_pronote_session = {
|
||
|
|
'url': qr_data.get('url'),
|
||
|
|
'token': _pronote_client.password, # Token en clair direct
|
||
|
|
'pin': pin,
|
||
|
|
'account_pin': account_pin,
|
||
|
|
'device_uuid': device_uuid,
|
||
|
|
'connected_at': datetime.now(timezone.utc)
|
||
|
|
}
|
||
|
|
|
||
|
|
# Sauvegarder en base
|
||
|
|
save_session_to_db(_pronote_session)
|
||
|
|
|
||
|
|
logger.info(f"Connexion PRONOTE réussie avec UUID: {device_uuid}")
|
||
|
|
return _pronote_client
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur connexion PRONOTE: {e}")
|
||
|
|
_pronote_client = None
|
||
|
|
_pronote_session = None
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def connect_with_credentials(url: str, username: str, password: str, account_pin: Optional[str] = None) -> Optional['PronoteClient']:
|
||
|
|
"""
|
||
|
|
Connecte à Pronote via identifiants classiques.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
url: URL du serveur Pronote
|
||
|
|
username: Nom d'utilisateur
|
||
|
|
password: Mot de passe
|
||
|
|
account_pin: PIN du compte (optionnel)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Client si connexion réussie, None sinon
|
||
|
|
"""
|
||
|
|
global _pronote_client, _pronote_session
|
||
|
|
|
||
|
|
try:
|
||
|
|
from .pronotepy import VieScolaireClient
|
||
|
|
|
||
|
|
_pronote_client = VieScolaireClient(
|
||
|
|
pronote_url=url,
|
||
|
|
username=username,
|
||
|
|
password=password,
|
||
|
|
mode="token",
|
||
|
|
uuid=str(uuid.uuid4()),
|
||
|
|
account_pin=account_pin
|
||
|
|
)
|
||
|
|
|
||
|
|
# Récupérer le token pour sauvegarde
|
||
|
|
token = _pronote_client.password if hasattr(_pronote_client, 'password') else None
|
||
|
|
device_uuid = str(_pronote_client.uuid) if hasattr(_pronote_client, 'uuid') else str(uuid.uuid4())
|
||
|
|
|
||
|
|
_pronote_session = {
|
||
|
|
'url': url,
|
||
|
|
'username': username,
|
||
|
|
'token': token,
|
||
|
|
'device_uuid': device_uuid,
|
||
|
|
'account_pin': account_pin,
|
||
|
|
'connected_at': datetime.now(timezone.utc)
|
||
|
|
}
|
||
|
|
|
||
|
|
# Sauvegarder en base
|
||
|
|
save_session_to_db(_pronote_session)
|
||
|
|
|
||
|
|
logger.info(f"Connexion PRONOTE réussie pour {username}")
|
||
|
|
return _pronote_client
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur connexion PRONOTE: {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def get_client() -> Optional['PronoteClient']:
|
||
|
|
"""
|
||
|
|
Retourne le client PRONOTE actuel s'il existe.
|
||
|
|
Tente de reconstituer la session depuis la base si le client est None.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
PronoteClient ou None
|
||
|
|
"""
|
||
|
|
global _pronote_client
|
||
|
|
|
||
|
|
if _pronote_client is not None:
|
||
|
|
return _pronote_client
|
||
|
|
|
||
|
|
# Essayer de reconstituer depuis la base
|
||
|
|
try:
|
||
|
|
from app_new.extensions import db
|
||
|
|
from app_new.pronote.models import PronoteSession
|
||
|
|
from .pronotepy import VieScolaireClient as PronoteClient
|
||
|
|
|
||
|
|
session = PronoteSession.get()
|
||
|
|
if session and session.session_token and session.session_url:
|
||
|
|
logger.info("Reconstruction session PRONOTE depuis la base")
|
||
|
|
# Reconstituer le client avec token_login
|
||
|
|
# L'ancien token devrait avoir été sauvegardé avec le UUID utilisé
|
||
|
|
token_uuid = session.device_uuid or str(uuid.uuid4())
|
||
|
|
_pronote_client = PronoteClient.token_login(
|
||
|
|
pronote_url=session.session_url,
|
||
|
|
username=session.username,
|
||
|
|
password=session.session_token,
|
||
|
|
uuid=token_uuid,
|
||
|
|
device_name='GMAO College',
|
||
|
|
account_pin=session.account_pin # Utiliser le PIN de la base
|
||
|
|
)
|
||
|
|
|
||
|
|
# Sauvegarder le nouveau token après reconnexion (le token peut tourner)
|
||
|
|
if _pronote_client:
|
||
|
|
new_token = _pronote_client.password if hasattr(_pronote_client, 'password') else None
|
||
|
|
if new_token and new_token != session.session_token:
|
||
|
|
logger.info("Mise à jour du token après reconnexion")
|
||
|
|
session.session_token = new_token
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
return _pronote_client
|
||
|
|
except Exception as e:
|
||
|
|
logger.warning(f"Impossible de reconstituer session: {e}")
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def get_session() -> Optional[Dict[str, Any]]:
|
||
|
|
"""
|
||
|
|
Retourne les informations de session actuelles.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict avec infos de session ou None
|
||
|
|
"""
|
||
|
|
return _pronote_session
|
||
|
|
|
||
|
|
|
||
|
|
def disconnect():
|
||
|
|
"""Déconnecte le client PRONOTE."""
|
||
|
|
global _pronote_client, _pronote_session
|
||
|
|
_pronote_client = None
|
||
|
|
_pronote_session = None
|
||
|
|
logger.info("Déconnexion PRONOTE")
|
||
|
|
|
||
|
|
|
||
|
|
def is_connected() -> bool:
|
||
|
|
"""Vérifie si un client pronote est connecté."""
|
||
|
|
return _pronote_client is not None
|
||
|
|
|
||
|
|
|
||
|
|
def get_all_salles(client) -> list:
|
||
|
|
"""
|
||
|
|
Récupère toutes les salles depuis Pronote.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
client: Client PRONOTE connecté
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Liste des salles avec leurs informations
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
# Utiliser l'API PRONOTE pour récupérer les ressources
|
||
|
|
result = client.post('RessourceEDT', 16, {})
|
||
|
|
ressources = result['dataSec']['data']['listeRessources']['V']
|
||
|
|
|
||
|
|
salles = []
|
||
|
|
for r in ressources:
|
||
|
|
if r.get('G') == 81: # Type salle = 81
|
||
|
|
salles.extend(r.get('liste', {}).get('V', []))
|
||
|
|
|
||
|
|
logger.info(f"Récupéré {len(salles)} salles depuis PRONOTE")
|
||
|
|
return salles
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur récupération salles: {e}")
|
||
|
|
return []
|
||
|
|
|
||
|
|
|
||
|
|
def get_lessons_for_room(client, room_name: str, date_start, date_end) -> list:
|
||
|
|
"""Récupère les cours pour une salle donnée en utilisant l'API directe."""
|
||
|
|
try:
|
||
|
|
# 1. Récupérer les salles
|
||
|
|
result = client.post('RessourceEDT', 16, {})
|
||
|
|
salles = []
|
||
|
|
for r in result['dataSec']['data']['listeRessources']['V']:
|
||
|
|
if r.get('G') == 81:
|
||
|
|
salles = r.get('liste', {}).get('V', [])
|
||
|
|
break
|
||
|
|
|
||
|
|
# 2. Trouver l'ID de la salle
|
||
|
|
salle_id = None
|
||
|
|
for s in salles:
|
||
|
|
if s.get('L') == room_name:
|
||
|
|
salle_id = s.get('N')
|
||
|
|
break
|
||
|
|
|
||
|
|
if not salle_id:
|
||
|
|
logger.warning(f"Salle {room_name} non trouvée dans Pronote")
|
||
|
|
return []
|
||
|
|
|
||
|
|
room_lessons = []
|
||
|
|
|
||
|
|
# 3. Pour chaque jour de la semaine
|
||
|
|
for d in range(7):
|
||
|
|
target_date = date_start + timedelta(days=d)
|
||
|
|
date_pronote = {"_T": 7, "V": f"{target_date.day}/{target_date.month}/{target_date.year} 0:0:0"}
|
||
|
|
|
||
|
|
try:
|
||
|
|
resultat = client.post('PageEmploiDuTemps', 16, {
|
||
|
|
"ressource": {"N": salle_id, "G": 17, "L": room_name},
|
||
|
|
"dateDebut": date_pronote,
|
||
|
|
"avecAbsencesEleve": False,
|
||
|
|
"avecConseilDeClasse": True,
|
||
|
|
"avecRetenuesEleve": True,
|
||
|
|
"estEDTPermanence": False,
|
||
|
|
"estEDTAnnuel": False,
|
||
|
|
"avecAbsencesRessource": True,
|
||
|
|
"avecCoursSortiePeda": True,
|
||
|
|
"avecDisponibilites": False,
|
||
|
|
"avecInfosPrefsGrille": True,
|
||
|
|
"edt": {"G": 81, "L": "Salles"},
|
||
|
|
"Ressource": {"N": salle_id, "G": 17, "L": room_name},
|
||
|
|
"DateDebut": date_pronote
|
||
|
|
})
|
||
|
|
except Exception as api_err:
|
||
|
|
logger.warning(f"Erreur API pour {room_name} le {target_date}: {api_err}")
|
||
|
|
continue
|
||
|
|
|
||
|
|
# 4. Parser les cours
|
||
|
|
liste_cours = resultat.get('dataSec', {}).get('data', {}).get('ListeCours', [])
|
||
|
|
if not liste_cours:
|
||
|
|
continue
|
||
|
|
|
||
|
|
for c in liste_cours:
|
||
|
|
statut = c.get('Statut', 'OK')
|
||
|
|
annule = c.get('estAnnule', False)
|
||
|
|
|
||
|
|
liste = c.get('ListeContenus', {}).get('V', [])
|
||
|
|
matiere = next((x.get('L') for x in liste if x.get('G') == 16), '')
|
||
|
|
classe = next((x.get('L') for x in liste if x.get('G') == 1), '')
|
||
|
|
prof = next((x.get('L') for x in liste if x.get('G') == 3), '')
|
||
|
|
|
||
|
|
# Parser l'heure - format "jj/mm/aaaa HH:MM"
|
||
|
|
heure_str = c.get('DateDuCours', {}).get('V', '')
|
||
|
|
|
||
|
|
# Calculer start_time et end_time
|
||
|
|
start_time = None
|
||
|
|
end_time = None
|
||
|
|
if ' ' in heure_str:
|
||
|
|
time_part = heure_str.split(' ')[-1]
|
||
|
|
try:
|
||
|
|
hours, minutes = map(int, time_part.split(':')[:2])
|
||
|
|
start_time = f"{hours:02d}:{minutes:02d}"
|
||
|
|
# Durée par défaut 1h, ou utiliser 'duree'
|
||
|
|
duree = c.get('duree', 1)
|
||
|
|
end_hours = hours + duree
|
||
|
|
end_time = f"{end_hours:02d}:{minutes:02d}"
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
room_lessons.append({
|
||
|
|
'day_of_week': d, # 0-6 (lundi=0)
|
||
|
|
'start_time': start_time,
|
||
|
|
'end_time': end_time,
|
||
|
|
'subject': matiere,
|
||
|
|
'teacher': prof,
|
||
|
|
'class_name': classe,
|
||
|
|
'statut': statut,
|
||
|
|
'annule': annule
|
||
|
|
})
|
||
|
|
|
||
|
|
logger.info(f"Récupéré {len(room_lessons)} cours pour salle {room_name}")
|
||
|
|
# Trier par jour puis par heure
|
||
|
|
room_lessons.sort(key=lambda x: (x['day_of_week'], x['start_time'] or ''))
|
||
|
|
return room_lessons
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur récupération cours pour {room_name}: {e}")
|
||
|
|
import traceback
|
||
|
|
logger.error(traceback.format_exc())
|
||
|
|
return []
|
||
|
|
|
||
|
|
|
||
|
|
# Fonction pour sauvegarder la session en base
|
||
|
|
def save_session_to_db(session_data: Dict[str, Any]):
|
||
|
|
"""
|
||
|
|
Sauvegarde les informations de session en base.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
session_data: Données de session à sauvegarder
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from app_new.extensions import db
|
||
|
|
from app_new.pronote.models import PronoteSession
|
||
|
|
|
||
|
|
# Supprimer les anciennes sessions
|
||
|
|
PronoteSession.query.delete()
|
||
|
|
|
||
|
|
# Créer nouvelle session
|
||
|
|
session = PronoteSession(
|
||
|
|
session_token=session_data.get('token'),
|
||
|
|
session_url=session_data.get('url'),
|
||
|
|
username=session_data.get('username'),
|
||
|
|
device_uuid=session_data.get('device_uuid'),
|
||
|
|
account_pin=session_data.get('account_pin'),
|
||
|
|
last_connection=datetime.now(timezone.utc)
|
||
|
|
)
|
||
|
|
db.session.add(session)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
logger.info("Session PRONOTE sauvegardée en base")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur sauvegarde session: {e}")
|
||
|
|
try:
|
||
|
|
db.session.rollback()
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def load_session_from_db() -> Optional[Dict[str, Any]]:
|
||
|
|
"""
|
||
|
|
Charge les informations de session depuis la base.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict avec infos de session ou None
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from app_new.pronote.models import PronoteSession
|
||
|
|
|
||
|
|
session = PronoteSession.get()
|
||
|
|
if session:
|
||
|
|
return {
|
||
|
|
'token': session.session_token,
|
||
|
|
'url': session.session_url,
|
||
|
|
'username': session.username,
|
||
|
|
'last_connection': session.last_connection
|
||
|
|
}
|
||
|
|
return None
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Erreur chargement session: {e}")
|
||
|
|
return None
|