#!/usr/bin/env python3 """Watchdog pour la synchronisation automatique des plannings PRONOTE - Pattern DND.""" import sys import os import time import signal from datetime import datetime, timedelta, timezone # Définir le répertoire de travail WORKDIR = os.environ.get('DOCKER_GMAO_WORKDIR', '/root/hermes-workspace/gmao-college') os.chdir(WORKDIR) sys.path.insert(0, WORKDIR) # Activer le venv venv_site_packages = os.path.join(WORKDIR, 'venv', 'lib', 'python3.13', 'site-packages') if os.path.exists(venv_site_packages): sys.path.insert(0, venv_site_packages) # Charger les variables d'environnement depuis .env from dotenv import load_dotenv env_file = os.environ.get('DOCKER_GMAO_ENVFILE', os.path.join(WORKDIR, '.env')) load_dotenv(env_file) # Fichier de log et de dernière exécution LOG_FILE = '/tmp/pronote_watchdog.log' LAST_RUN_FILE = '/tmp/pronote_last_run.txt' running = True def log(message, level='info'): """Écrit un message dans le log et dans la base centralisee.""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f'[{timestamp}] {message}') with open(LOG_FILE, 'a') as f: f.write(f'[{timestamp}] {message}\n') try: sys.path.insert(0, '/app') from app_new.watchdog_logger import log_watchdog log_watchdog('pronote_watchdog', message, level=level) except Exception: pass def get_last_run(): """Retourne la date de dernière exécution.""" if os.path.exists(LAST_RUN_FILE): try: with open(LAST_RUN_FILE, 'r') as f: return datetime.fromisoformat(f.read().strip()) except: return None return None def save_last_run(): """Sauvegarde la date de dernière exécution.""" with open(LAST_RUN_FILE, 'w') as f: f.write(datetime.now().isoformat()) def should_run(gmao_context): """Vérifie si le watchdog doit s'exécuter selon l'intervalle configuré.""" if not gmao_context or not gmao_context.pronote_sync_enabled: return False last_run = get_last_run() if not last_run: return True interval_minutes = gmao_context.pronote_sync_interval or 360 next_run = last_run + timedelta(minutes=interval_minutes) return datetime.now() >= next_run def sync_pronote_plannings(): """Synchronise les plannings de toutes les salles depuis PRONOTE.""" from app_new import create_app, db from app_new.outlook.models import GmaoContext from app_new.core.models.college import Room, RoomSchedule, RoomType from app_new.lib_ext.pronote_client import get_client, get_lessons_for_room app = create_app() with app.app_context(): gmao_context = GmaoContext.get_for_user(2) # TODO: rendre configurable weeks = gmao_context.pronote_sync_weeks if gmao_context else 4 # Récupérer le client PRONOTE client = get_client() if not client: log("ERREUR: Client PRONOTE non connecté") return {'success': False, 'error': 'Client PRONOTE non connecté'} # Récupérer toutes les salles d'enseignement (via RoomType) teaching_rooms = Room.query.join(RoomType).filter(RoomType.is_teaching == True).all() if not teaching_rooms: log("Aucune salle d'enseignement trouvée") return {'success': True, 'rooms': 0, 'schedules': 0} log(f"Synchronisation de {len(teaching_rooms)} salles sur {weeks} semaines") total_schedules = 0 rooms_updated = 0 errors = [] not_found = [] today = datetime.now().date() for room in teaching_rooms: try: room_schedules_count = 0 # Pour chaque semaine (0 à weeks-1) for week_offset in range(weeks): # Calculer le lundi de la semaine days_since_monday = today.weekday() monday = today - timedelta(days=days_since_monday) + timedelta(weeks=week_offset) sunday = monday + timedelta(days=6) # Récupérer les cours pour cette salle lessons = get_lessons_for_room(client, room.name, monday, sunday) if not lessons: continue # Supprimer les anciens plannings pour cette période RoomSchedule.query.filter( RoomSchedule.room_id == room.id, RoomSchedule.week_start == monday ).delete() # Insérer les nouveaux plannings for lesson in lessons: # Ignorer les cours sans horaires if not lesson.get('start_time') or not lesson.get('end_time'): continue # Convertir les strings en objets time try: start_time_str = lesson.get('start_time') end_time_str = lesson.get('end_time') h_start, m_start = map(int, start_time_str.split(':')) h_end, m_end = map(int, end_time_str.split(':')) start_time = datetime.min.replace(hour=h_start, minute=m_start).time() end_time = datetime.min.replace(hour=h_end, minute=m_end).time() except: continue schedule = RoomSchedule( room_id=room.id, week_start=monday, day_of_week=lesson.get('day_of_week', 0), start_time=start_time, end_time=end_time, subject=lesson.get('subject'), teacher=lesson.get('teacher'), class_name=lesson.get('class_name'), course_name=lesson.get('subject') ) db.session.add(schedule) total_schedules += 1 room_schedules_count += 1 rooms_updated += 1 if room_schedules_count > 0: log(f" ✓ {room.name}: {room_schedules_count} cours") else: not_found.append(room.name) except Exception as e: errors.append(f"{room.name}: {str(e)}") log(f" ⚠ Erreur salle {room.name}: {str(e)}") # Pause d'une minute entre chaque salle pour éviter le blocage IP if room != teaching_rooms[-1]: # Pas de pause après la dernière salle log(f" Pause 60s avant la prochaine salle...") time.sleep(60) db.session.commit() # Mettre à jour la date de dernière connexion PRONOTE try: from app_new.pronote.models import PronoteSession session = PronoteSession.get() if session: session.last_connection = datetime.now(timezone.utc).replace(tzinfo=None) db.session.commit() except Exception as e: log(f"Erreur mise à jour last_connection PRONOTE: {e}", level='error') if not_found: log(f" {len(not_found)} salles sans cours dans PRONOTE: {', '.join(not_found[:5])}{'...' if len(not_found) > 5 else ''}") return { 'success': True, 'rooms': rooms_updated, 'schedules': total_schedules, 'errors': errors, 'not_found': len(not_found) } def run_watchdog(): """Exécute un cycle du watchdog - retourne False si désactivé.""" global running from app_new import create_app, db from app_new.outlook.models import GmaoContext app = create_app() with app.app_context(): # Récupérer le contexte GMAO gmao_context = GmaoContext.get_for_user(2) # TODO: rendre configurable # Vérifier si activé if not gmao_context or not gmao_context.pronote_sync_enabled: log("Watchdog PRONOTE désactivé dans la config") return False # Arrêter le watchdog if not should_run(gmao_context): # Pas le moment, attendre return True interval = gmao_context.pronote_sync_interval or 360 weeks = gmao_context.pronote_sync_weeks or 4 log(f"Exécution synchronisation PRONOTE (intervalle: {interval} min, semaines: {weeks})...") try: result = sync_pronote_plannings() if result.get('success'): log(f"✓ {result.get('rooms', 0)} salles, {result.get('schedules', 0)} plannings") else: log(f"ERREUR: {result.get('error', 'inconnue')}") if result.get('errors'): for err in result.get('errors', []): log(f" ⚠ {err[:100]}") save_last_run() db.session.commit() except Exception as e: db.session.rollback() import traceback log(f"ERREUR dans le cycle: {str(e)}") log(f"Traceback: {traceback.format_exc()[:500]}") try: db.session.rollback() except: pass return True # Continuer def main(): """Point d'entrée principal - pattern DND avec redémarrage automatique.""" global running log("="*50) log("Watchdog PRONOTE démarré (pattern DND)") log("="*50) # Capturer le signal d'arrêt def signal_handler(signum, frame): global running running = False log("Signal reçu, arrêt du watchdog...") signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler) while running: try: result = run_watchdog() if result is False: log("Watchdog désactivé - arrêt") break # Sortir proprement si désactivé except Exception as e: import traceback log(f"Crash: {str(e)}", level='error') log(f"Traceback: {traceback.format_exc()[-500:]}", level='error') # Redémarrer automatiquement après crash # Attendre avant la prochaine vérification # Si le watchdog est désactivé, on attend plus longtemps pour éviter les redémarrages incessants wait_seconds = 300 if result is False else 30 for _ in range(wait_seconds): if not running: break time.sleep(1) log("Watchdog PRONOTE arrêté") if __name__ == '__main__': main()