from __future__ import annotations from datetime import datetime import logging import traceback import requests from .config import AppConfig from .db import get_db_connection from .email_alerts import send_alert_email from .parsers import etat_to_int, parse_date_only, parse_iso_datetime, toxines_to_str from .repository import fetch_sites_mapping, insert_history_if_new, update_site_current_state def fetch_api_data(api_url: str) -> dict: headers = {"User-Agent": "Mozilla/5.0 (compatible; import_situations_api/1.2)"} response = requests.get(api_url, headers=headers, timeout=30) response.raise_for_status() return response.json() def run_import(config: AppConfig, logger: logging.Logger) -> None: start_time = datetime.now() updated_count = 0 historised_count = 0 skipped_no_match = 0 conn = None logger.info("Démarrage du script import_situations_api") try: try: conn = get_db_connection(config.env, logger) except Exception: logger.error("Impossible de se connecter à la base, arrêt.") return try: data = fetch_api_data(config.api_url) except Exception as exc: logger.error("Erreur lors de l'appel à l'API : %s", exc) logger.debug("Traceback : %s", traceback.format_exc()) return data_actualizacion = data.get("dataActualizacion") date_document, heure_document = parse_iso_datetime(data_actualizacion, logger) logger.info( "API récupérée - dataActualizacion=%s (date_document=%s, heure_document=%s)", data_actualizacion, date_document, heure_document, ) estado_zonas = data.get("estadoZonas", []) if not estado_zonas: logger.warning("Aucune entrée 'estadoZonas' dans la réponse API.") return mapping = fetch_sites_mapping(conn) logger.info("Mapping sites chargé : %d noms_poligono distincts", len(mapping)) state_changes: list[dict] = [] for item in estado_zonas: if item.get("tipo", "") != "BI_CV": continue nome = (item.get("nome") or "").strip() if not nome: continue sites_for_nome = mapping.get(nome) if not sites_for_nome: skipped_no_match += 1 continue estado_txt = item.get("estado", "") estado_detalle = item.get("estadoDetalle") data_sit_txt = item.get("dataSituacionAdministrativa") plan = (estado_detalle or "").strip() or None toxines = toxines_to_str(item.get("toxicidade")) etat_admin_int = etat_to_int(estado_txt) date_changement = parse_date_only(data_sit_txt, logger) for site_row in sites_for_nome: id_site = site_row["id"] nom_site = site_row["nom"] nom_poligono = site_row["nom_poligono"] logger.info( "Mise à jour site id=%s (nom='%s', nom_poligono='%s') -> etat=%s, plan=%s, toxines=%s, date_changement=%s", id_site, nom_site, nom_poligono, estado_txt, plan, toxines, date_changement, ) etat_precedent = site_row.get("etat_administratif") if etat_precedent is not None and int(etat_precedent) != etat_admin_int: state_changes.append({ "nom_site": nom_site, "nom_poligono": nom_poligono, "etat_avant": int(etat_precedent), "etat_apres": etat_admin_int, "plan": plan, "toxines": toxines, "date_changement": date_changement, }) try: update_site_current_state( conn, logger, id_site=id_site, plan=plan, toxines=toxines, etat_administratif=etat_admin_int, date_changement_etat=date_changement, ) updated_count += 1 insert_history_if_new( conn, logger, id_site=id_site, date_document=date_document, heure_document=heure_document, plan=plan, toxines=toxines, etat_administratif=etat_admin_int, date_changement_etat=date_changement, ) historised_count += 1 except Exception as exc: logger.error( "Erreur lors de la mise à jour/historisation du site id=%s : %s", id_site, exc, ) logger.debug("Traceback : %s", traceback.format_exc()) try: conn.commit() except Exception as exc: logger.error("Erreur lors du commit : %s", exc) logger.debug("Traceback : %s", traceback.format_exc()) if state_changes: send_alert_email(config.env, logger, state_changes) logger.info( "Traitement terminé : %d mises à jour, %d tentatives d'historisation, %d zones sans correspondance", updated_count, historised_count, skipped_no_match, ) except Exception as exc: logger.error("Erreur inattendue dans import_situations_api : %s", exc) logger.error(traceback.format_exc()) finally: if conn is not None: try: conn.close() except Exception: pass end_time = datetime.now() duration = (end_time - start_time).total_seconds() logger.info( "Fin du script import_situations_api - début=%s fin=%s durée=%.2f s", start_time.strftime("%Y-%m-%d %H:%M:%S"), end_time.strftime("%Y-%m-%d %H:%M:%S"), duration, ) logger.warning( "import_situations_api - Résumé : debut=%s fin=%s durée=%.2fs MAJ=%d HISTO=%d SKIP=%d", start_time.strftime("%Y-%m-%d %H:%M:%S"), end_time.strftime("%Y-%m-%d %H:%M:%S"), duration, updated_count, historised_count, skipped_no_match, ) def test_email(config: AppConfig, logger: logging.Logger) -> int: alert_to_raw = config.env.get("ALERT_TO", "") smtp_host = config.env.get("SMTP_HOST", "") print(f"SMTP_HOST : {smtp_host or '(non configuré)'}") print(f"SMTP_PORT : {config.env.get('SMTP_PORT', '587')}") print(f"SMTP_USER : {config.env.get('SMTP_USER', '(non configuré)')}") print(f"ALERT_FROM : {config.env.get('ALERT_FROM', '(non configuré)')}") print(f"ALERT_TO : {alert_to_raw or '(non configuré)'}") print() if not smtp_host or not alert_to_raw: print("[ERREUR] SMTP_HOST ou ALERT_TO manquant dans .env - impossible d'envoyer.") return 1 fake_changes = [ { "nom_site": "Sada 1", "nom_poligono": "Sada 1", "etat_avant": 0, "etat_apres": 1, "plan": "B1", "toxines": "Lipofilicas,ASP", "date_changement": datetime.now().date(), }, { "nom_site": "Cambados 3", "nom_poligono": "Cambados 3", "etat_avant": 1, "etat_apres": 0, "plan": None, "toxines": None, "date_changement": datetime.now().date(), }, ] print("Envoi de l'email de test avec 2 cas : un site ferme et un site ouvert...") send_alert_email(config.env, logger, fake_changes) print("Termine - verifiez votre boite mail.") return 0