566 lines
20 KiB
Python
566 lines
20 KiB
Python
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) |