import json import re import os import time import base64 import configparser import requests from datetime import datetime, timedelta from zoneinfo import ZoneInfo from playwright.sync_api import sync_playwright from PIL import Image, ImageDraw, ImageFont import websocket # Nécessite: pip install websocket-client # === Configuration (config.ini en dev, variables d'environnement en conteneur) === CONFIG_FILE = "config.ini" HEALTH_FILE = "/tmp/health_status.json" USE_ENV = os.getenv("USE_ENV", "false").lower() in ("true", "1", "yes") or not os.path.exists(CONFIG_FILE) if not USE_ENV: print(f"🔧 Mode DEV local : Chargement de la configuration depuis '{CONFIG_FILE}'") config = configparser.ConfigParser() config.read(CONFIG_FILE, encoding="utf-8") HA_URL = config.get("HOME_ASSISTANT", "url").rstrip('/') HA_TOKEN = config.get("HOME_ASSISTANT", "token") HA_WEBHOOK_ID = config.get("HOME_ASSISTANT", "webhook_id") HA_CALENDAR_ID = config.get("HOME_ASSISTANT", "calendar_id") BASE_URL = config.get("CCI", "base_url").rstrip('/') CCI_USERNAME = config.get("CCI", "username") CCI_PASSWORD = config.get("CCI", "password") CODE_RESSOURCE = config.getint("CCI", "code_ressource") TYPE_RESSOURCE = config.getint("CCI", "type_ressource") STATE_FILE = config.get("SCRIPT", "state_file", fallback="planning_state.json") SCREENSHOT_DIR = config.get("SCRIPT", "screenshot_dir", fallback="screenshots") CHECK_INTERVAL = config.getint("SCRIPT", "check_interval", fallback=3600) # Gestion propre du fuseau horaire TZ = ZoneInfo(config.get("SCRIPT", "tz", fallback="Europe/Paris")) else: print("🐳 Mode CONTENEUR (Docker) : Chargement de la configuration depuis les variables d'environnement") HA_URL = os.getenv("HA_URL", "").rstrip('/') HA_TOKEN = os.getenv("HA_TOKEN", "") HA_WEBHOOK_ID = os.getenv("HA_WEBHOOK_ID", "") HA_CALENDAR_ID = os.getenv("HA_CALENDAR_ID", "") BASE_URL = os.getenv("CCI_BASE_URL", "").rstrip('/') CCI_USERNAME = os.getenv("CCI_USERNAME", "") CCI_PASSWORD = os.getenv("CCI_PASSWORD", "") CODE_RESSOURCE = int(os.getenv("CCI_CODE_RESSOURCE", "0")) TYPE_RESSOURCE = int(os.getenv("CCI_TYPE_RESSOURCE", "0")) STATE_FILE = os.getenv("STATE_FILE", "planning_state.json") SCREENSHOT_DIR = os.getenv("SCREENSHOT_DIR", "screenshots") CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "3600")) # Gestion propre du fuseau horaire TZ = ZoneInfo(os.getenv("TZ", "Europe/Paris")) required_vars = [ ("HA_URL", HA_URL), ("HA_TOKEN", HA_TOKEN), ("HA_WEBHOOK_ID", HA_WEBHOOK_ID), ("HA_CALENDAR_ID", HA_CALENDAR_ID), ("CCI_BASE_URL", BASE_URL), ("CCI_USERNAME", CCI_USERNAME), ("CCI_PASSWORD", CCI_PASSWORD), ] missing = [name for name, val in required_vars if not val] if missing: raise ValueError(f"❌ Configuration incomplète. Paramètres/Variables manquants : {', '.join(missing)}") if not os.path.exists(SCREENSHOT_DIR): os.makedirs(SCREENSHOT_DIR, exist_ok=True) def update_health_status(status="ok", error=None): """Met à jour le fichier d'état de santé pour le healthcheck Docker.""" data = { "status": status, "timestamp": time.time(), "last_check": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "error": error } try: with open(HEALTH_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception as e: print(f"⚠️ Erreur mise à jour fichier health: {e}") def get_weeks_2026_2027(): """Génère la liste des semaines au format YYYYSS de sept 2026 à août 2027.""" weeks = [] current_date = datetime(2026, 9, 1) end_date = datetime(2027, 8, 31) while current_date <= end_date: year, week_num, _ = current_date.isocalendar() if current_date.month == 12 and week_num == 1: year += 1 elif current_date.month == 1 and week_num > 50: year -= 1 week_str = f"{year}{week_num:02d}" if week_str not in weeks: weeks.append(week_str) current_date += timedelta(days=7) return weeks def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) return {} def save_state(state): state_dir = os.path.dirname(STATE_FILE) if state_dir: os.makedirs(state_dir, exist_ok=True) with open(STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def timestamp_image(image_path): """Ajoute la date et l'heure en filigrane sur la capture d'écran.""" img = Image.open(image_path) draw = ImageDraw.Draw(img) try: font = ImageFont.load_default(size=28) except: font = ImageFont.load_default() timestamp = datetime.now().strftime("Capture du : %d/%m/%Y à %H:%M:%S") draw.rectangle([(10, 10), (460, 50)], fill="black") draw.text((20, 15), timestamp, fill="white", font=font) img.save(image_path) def handle_week_screenshot(page, week, is_first_time, week_changed): """Gère la capture d'écran par semaine.""" week_dir = os.path.join(SCREENSHOT_DIR, f"semaine_{week}") os.makedirs(week_dir, exist_ok=True) latest_filename = f"latest.png" latest_path = os.path.join(week_dir, latest_filename) temp_path = os.path.join(week_dir, f"temp_{week}.png") page.screenshot(path=temp_path, full_page=True) timestamp_image(temp_path) if week_changed and not is_first_time and os.path.exists(latest_path): mtime = os.path.getmtime(latest_path) dt_mtime = datetime.fromtimestamp(mtime) timestamp_str = dt_mtime.strftime("%Y-%m-%d_%H-%M-%S") archive_filename = f"{timestamp_str}.png" archive_path = os.path.join(week_dir, archive_filename) os.rename(latest_path, archive_path) print(f" 📸 Modification détectée : Ancien screenshot archivé -> {archive_filename}") if os.path.exists(latest_path): os.remove(latest_path) os.rename(temp_path, latest_path) print(f" 📸 Capture 'latest' mise à jour : {latest_path}") def notify_home_assistant(date_str, before, after): """Envoie un Webhook à Home Assistant avec les données formatées.""" try: dt_obj = datetime.strptime(date_str, "%d/%m/%Y") date_iso = dt_obj.strftime("%Y-%m-%d") except ValueError: date_iso = date_str url = f"{HA_URL}/api/webhook/{HA_WEBHOOK_ID}" payload = { "action": "planning_updated", "date_iso": date_iso, # ex: "2026-09-14" "date_humain": date_str, # ex: "14/09/2026" "before": before, # Séances avant modif (sans numJour, numSemaine, minuteDebut) "after": after # Séances après modif (sans numJour, numSemaine, minuteDebut) } try: resp = requests.post(url, json=payload, timeout=10) if resp.status_code == 200: print(f" 📡 Webhook HA envoyé pour le {date_str} ({date_iso})") else: print(f" ⚠️ Webhook HA a renvoyé le statut {resp.status_code}") except Exception as e: print(f" ❌ Erreur Webhook HA: {e}") def get_ha_events_for_day(date_str): """Récupère les événements existants dans HA pour une date (DD/MM/YYYY).""" try: dt_start = datetime.strptime(date_str, "%d/%m/%Y") dt_end = dt_start + timedelta(days=1) start_iso = dt_start.strftime("%Y-%m-%dT00:00:00Z") end_iso = dt_end.strftime("%Y-%m-%dT00:00:00Z") url = f"{HA_URL}/api/calendars/{HA_CALENDAR_ID}?start={start_iso}&end={end_iso}" headers = {"Authorization": f"Bearer {HA_TOKEN}"} resp = requests.get(url, headers=headers, timeout=10) if resp.status_code == 200: return resp.json() except Exception as e: print(f" ⚠️ Impossible de lire le calendrier HA pour le {date_str} : {e}") return [] def delete_ha_calendar_event(ha_url, ha_token, calendar_entity_id, uid): """Supprime un événement dans Home Assistant via WebSocket.""" if not uid: print(" ❌ Aucun UID fourni pour la suppression.") return False ws_url = ha_url.replace("https://", "wss://").replace("http://", "ws://").rstrip('/') + "/api/websocket" try: ws = websocket.create_connection(ws_url, timeout=10) init_response = json.loads(ws.recv()) if init_response.get("type") != "auth_required": ws.close() return False ws.send(json.dumps({ "type": "auth", "access_token": ha_token })) auth_response = json.loads(ws.recv()) if auth_response.get("type") != "auth_ok": ws.close() return False delete_command = { "id": 1, "type": "calendar/event/delete", "entity_id": calendar_entity_id, "uid": uid } ws.send(json.dumps(delete_command)) result = json.loads(ws.recv()) ws.close() return result.get("success", False) except Exception as e: print(f" ❌ Erreur de connexion WebSocket : {e}") return False def create_ha_calendar_event(seance): """Crée un événement dans le calendrier Home Assistant en utilisant les ISO formatés.""" headers = { "Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json" } details = "\n".join(seance.get("detail", [])) if isinstance(seance.get("detail"), list) else str(seance.get("detail", "")) payload = { "entity_id": HA_CALENDAR_ID, "summary": seance["libelle"], "start_date_time": seance["debut_iso"], "end_date_time": seance["fin_iso"], "description": details } try: url = f"{HA_URL}/api/services/calendar/create_event" resp = requests.post(url, json=payload, headers=headers, timeout=10) if resp.status_code == 200: print(f" ✅ Cours créé : {seance['libelle']} ({seance['heure_debut']} - {seance['heure_fin']})") return True else: print(f" ⚠️ Erreur création HA ({resp.status_code}) : {resp.text}") return False except Exception as e: print(f" ❌ Erreur réseau lors de la création d'événement HA : {e}") return False def reconcile_day_ha_calendar(date_str, expected_seances): """Réconcilie les événements HA avec le planning CCI.""" print(f" 🔍 Audit et réconciliation du calendrier HA pour le {date_str}...") existing_events = get_ha_events_for_day(date_str) expected_matches = [] for seance in expected_seances: expected_matches.append({ "seance": seance, "summary": seance["libelle"], "heure_debut": seance["heure_debut"], "found": False }) events_to_delete = [] for ev in existing_events: ev_summary = ev.get("summary", "") ev_start_raw = ev.get("start", {}) ev_start_str = ev_start_raw.get("dateTime", "") if isinstance(ev_start_raw, dict) else str(ev_start_raw) ev_uid = ev.get("uid") or ev.get("id") matched = False for exp in expected_matches: if exp["found"]: continue if ev_summary == exp["summary"] and exp["heure_debut"] in ev_start_str: exp["found"] = True matched = True break if not matched: events_to_delete.append({ "summary": ev_summary, "uid": ev_uid, "start": ev_start_str }) if events_to_delete: for ev_del in events_to_delete: print(f" 🗑️ [Correction HA] Événement en trop détecté : {ev_del['summary']} ({ev_del['start']})") if ev_del["uid"]: if delete_ha_calendar_event(HA_URL, HA_TOKEN, HA_CALENDAR_ID, ev_del["uid"]): print(f" ✅ Événement supprimé du calendrier HA.") else: print(f" ⚠️ Échec de suppression de l'événement (UID: {ev_del['uid']}).") else: print(f" ⚠️ Impossible de supprimer '{ev_del['summary']}' : aucun UID renvoyé par l'API HA.") else: print(f" ✅ Aucun événement obsolète/en trop dans HA.") missing_count = 0 for exp in expected_matches: if not exp["found"]: missing_count += 1 print(f" ✨ [Correction HA] Événement manquant dans HA : {exp['summary']} ({exp['heure_debut']})") create_ha_calendar_event(exp["seance"]) if missing_count == 0: print(f" ✅ Tous les cours attendus sont bien présents dans HA.") def format_seance(seance, dt_debut_semaine): """ Transforme une séance brute en supprimant numJour/numSemaine/minuteDebut et en ajoutant date_iso, date_humain, debut_iso, fin_iso, heure_debut, heure_fin. """ num_jour = seance.get("numJour", 1) jour_delta = num_jour - 1 dt_jour = dt_debut_semaine + timedelta(days=jour_delta) minute_debut = seance.get("minuteDebut", 0) duree = seance.get("duree", 0) minute_fin = minute_debut + duree heure_debut_h = minute_debut // 60 heure_debut_m = minute_debut % 60 heure_fin_h = minute_fin // 60 heure_fin_m = minute_fin % 60 dt_start_naive = datetime(dt_jour.year, dt_jour.month, dt_jour.day, heure_debut_h, heure_debut_m) dt_end_naive = datetime(dt_jour.year, dt_jour.month, dt_jour.day, heure_fin_h, heure_fin_m) dt_start = dt_start_naive.replace(tzinfo=TZ) dt_end = dt_end_naive.replace(tzinfo=TZ) # Dégager numJour, numSemaine et minuteDebut cleaned = { k: v for k, v in seance.items() if k not in ("numJour", "numSemaine", "minuteDebut") } date_iso = dt_jour.strftime("%Y-%m-%d") date_humain = dt_jour.strftime("%d/%m/%Y") cleaned.update({ "date_iso": date_iso, "date_humain": date_humain, "debut_iso": dt_start.isoformat(), "fin_iso": dt_end.isoformat(), "heure_debut": f"{heure_debut_h:02d}:{heure_debut_m:02d}", "heure_fin": f"{heure_fin_h:02d}:{heure_fin_m:02d}", "duree_minutes": duree }) return cleaned, date_humain def parse_planning_to_days(planning_data): """Convertit le JSON brut en dictionnaire classé par date avec données formatées.""" days_data = {} seen_seances = set() for semaine in planning_data.get("semaines", []): date_debut_str = semaine.get("dateDebut") if not date_debut_str: continue try: dt_debut = datetime.strptime(date_debut_str, "%d/%m/%Y") except ValueError: continue for ressource in semaine.get("ressources", []): for seance in ressource.get("seances", []): seance_key = ( seance.get("id"), seance.get("numJour"), seance.get("minuteDebut"), seance.get("libelle") ) if seance_key in seen_seances: continue seen_seances.add(seance_key) formatted_seance, date_str = format_seance(seance, dt_debut) if date_str not in days_data: days_data[date_str] = [] days_data[date_str].append(formatted_seance) for date_str in days_data: days_data[date_str] = sorted(days_data[date_str], key=lambda x: x["debut_iso"]) return days_data def run(): weeks_to_check = get_weeks_2026_2027() state = load_state() ressources_json = json.dumps([{"code": CODE_RESSOURCE, "type": TYPE_RESSOURCE}]) ressources_b64 = base64.b64encode(ressources_json.encode()).decode() with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context(viewport={"width": 1920, "height": 1080}) page = context.new_page() print("🔑 Connexion au portail CCI...") page.goto(f"{BASE_URL}/IntNum/index.php") username_field = page.locator('input[type="text"], input[name="username"], input[name="login"]') username_field.click() username_field.press_sequentially(CCI_USERNAME, delay=20) password_field = page.locator('input[type="password"], input[name="password"]') password_field.click() password_field.press_sequentially(CCI_PASSWORD, delay=20) page.locator('body').click() page.wait_for_load_state("networkidle") print("✅ Connexion réussie !\n") for week in weeks_to_check: print("=" * 60) print(f"🔍 Traitement de la semaine {week}...") params = f"orientationPaysage=0&afficheContraintes=1&zoom=2&modeAffichage=0&semaineDebut={week}&semaineFin={week}&ressources={ressources_b64}" url = f"{BASE_URL}/IntNum/index.php/apprenant/planning/courant/?{params}" page.goto(url) page.wait_for_load_state("networkidle") page.wait_for_timeout(1000) content = page.content() match = re.search(r'var planningJSON\s*=\s*(\{.*?\});\s*\n', content, re.DOTALL) if not match: print(f"⚠️ Impossible d'extraire planningJSON pour la semaine {week}") continue try: planning_data = json.loads(match.group(1)) except Exception as e: print(f"⚠️ Erreur de lecture du JSON ({e})") continue new_days_data = parse_planning_to_days(planning_data) old_days_data = state.get(week, {}) is_first_time = (week not in state) week_changed = (new_days_data != old_days_data) handle_week_screenshot(page, week, is_first_time, week_changed) if not new_days_data: print(f"ℹ️ Aucun cours prévu cette semaine.") if is_first_time or week_changed: state[week] = {} save_state(state) print(f"✅ Semaine {week} vérifiée.") continue if is_first_time: print(f"🆕 Premier traitement pour cette semaine.") elif week_changed: print(f"🔄 Modifications détectées dans le planning CCI !") else: print(f"ℹ️ Planning CCI inchangé. Réconciliation du calendrier HA...") all_dates = sorted(list(set(new_days_data.keys()).union(set(old_days_data.keys())))) for date_str in all_dates: old_day = old_days_data.get(date_str, []) new_day = new_days_data.get(date_str, []) if old_day != new_day and not is_first_time: print(f" ⚠️ Changement détecté sur le planning pour la journée du {date_str}") notify_home_assistant(date_str, old_day, new_day) reconcile_day_ha_calendar(date_str, new_day) state[week] = new_days_data save_state(state) print(f"✅ Semaine {week} traitée et réconciliée avec succès.") browser.close() print("\n" + "=" * 60) print("🎉 Traitement et audit complets terminés avec succès !") if __name__ == "__main__": print(f"🔄 Service démarré. Intervalle d'analyse : {CHECK_INTERVAL} secondes ({CHECK_INTERVAL // 60} minutes)") while True: start_time = datetime.now() print("\n" + "=" * 60) print(f"🚀 Début du cycle d'analyse : {start_time.strftime('%d/%m/%Y %H:%M:%S')}") update_health_status(status="running") try: run() update_health_status(status="ok") except Exception as e: print(f"❌ Erreur lors du cycle d'exécution : {e}") update_health_status(status="error", error=str(e)) next_run = datetime.now() + timedelta(seconds=CHECK_INTERVAL) print(f"💤 Prochaine analyse prévue à {next_run.strftime('%H:%M:%S')} (dans {CHECK_INTERVAL}s)") print("=" * 60) time.sleep(CHECK_INTERVAL)