#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Script : import_situations_api.py Chemin : /var/www/html/analyses/import_situations_api.py Description : Récupère les situations administratives des zones de production via l'API INTECMAR : https://www.intecmar.gal/Api/EstadoZonasProducion/EstadoZonasBiotoxinas puis met à jour : - la table courante `analyses_sites` : * plan (A, B1, B2, B3, C1, C2, C3, D, ...) * toxines (liste jointe par virgules ex: "Lipofílicas,ASP") * etat_administratif (0=ouvert, 1=fermé) * date_changement_etat - la table d'historique `analyses_sites_historique` : * 1 ligne par changement (id_site + date_changement_etat) * avec date_document / heure_document de l'API Le lien se fait sur le champ `analyses_sites.nom_poligono` <-> `estadoZonas[].nome`. Options : aucune (exécution simple, prévue pour un cron) Exemple : python3 import_situations_api.py Prérequis : - Python 3 - Modules : requests, mysql-connector-python - Fichier .env dans le même répertoire avec : DB_HOST=localhost DB_PORT=3306 DB_USER=... DB_PASSWORD=... DB_NAME=analyses LOG_MODE=normal (ou debug) Auteur : Sylvain SCATTOLINI Date : 02/12/2025 Version : 1.2 """ import os import sys import json import logging import traceback from pathlib import Path from datetime import datetime import requests import mysql.connector # --------------------------------------------------------------------------- # Configuration de base # --------------------------------------------------------------------------- BASE_DIR = Path(__file__).resolve().parent ENV_PATH = BASE_DIR / ".env" LOG_DIR = BASE_DIR / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) LOG_FILE = LOG_DIR / "import_situations_api.log" API_URL = "https://www.intecmar.gal/Api/EstadoZonasProducion/EstadoZonasBiotoxinas" # --------------------------------------------------------------------------- # Chargement .env minimaliste # --------------------------------------------------------------------------- def load_env(env_path: Path) -> dict: env = {} if not env_path.is_file(): return env with env_path.open("r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) env[k.strip()] = v.strip() return env ENV = load_env(ENV_PATH) # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- def setup_logging(): logger = logging.getLogger("import_situations") logger.setLevel(logging.DEBUG) logger.handlers.clear() # ------------------------------------------------------------------ # LOG 1 : logs détaillés -> écrasés à chaque exécution # ------------------------------------------------------------------ detailed_log_path = LOG_DIR / "import_situations_api.log" fh = logging.FileHandler(detailed_log_path, mode="w", encoding="utf-8") fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" )) logger.addHandler(fh) # ------------------------------------------------------------------ # LOG 2 : intecmar.log -> uniquement erreurs + warning + résumé # ------------------------------------------------------------------ essential_log_path = LOG_DIR / "intecmar.log" eh = logging.FileHandler(essential_log_path, mode="a", encoding="utf-8") eh.setLevel(logging.WARNING) # WARNING + ERROR seulement eh.setFormatter(logging.Formatter( "%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S" )) logger.addHandler(eh) # ------------------------------------------------------------------ # Console (stdout) # ------------------------------------------------------------------ ch = logging.StreamHandler(sys.stdout) ch.setLevel(logging.INFO) ch.setFormatter(logging.Formatter( "%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S" )) logger.addHandler(ch) return logger logger = setup_logging() # --------------------------------------------------------------------------- # Connexion MySQL # --------------------------------------------------------------------------- def get_db_connection(): try: conn = mysql.connector.connect( host=ENV.get("DB_HOST", "localhost"), port=int(ENV.get("DB_PORT", "3306")), user=ENV.get("DB_USER", "root"), password=ENV.get("DB_PASSWORD", ""), database=ENV.get("DB_NAME", "analyses"), charset="utf8mb4", use_unicode=True, ) return conn except Exception as e: logger.error("Erreur de connexion MySQL : %s", e) raise # --------------------------------------------------------------------------- # Helpers de conversion # --------------------------------------------------------------------------- def parse_iso_datetime(dt_str: str): """ Convertit une chaîne ISO (ex: '2025-12-02T09:28:00+01:00') en (date, time) en ignorant le fuseau. """ if not dt_str: return None, None try: # On ignore le fuseau si présent if "+" in dt_str: dt_str = dt_str.split("+", 1)[0] if "Z" in dt_str: dt_str = dt_str.rstrip("Z") dt = datetime.fromisoformat(dt_str) return dt.date(), dt.time().replace(microsecond=0) except Exception: logger.warning("Impossible de parser la date ISO : %s", dt_str) return None, None def parse_date_only(dt_str: str): """ Convertit '2025-11-24T00:00:00' en date uniquement. """ if not dt_str: return None try: if "T" in dt_str: dt_str = dt_str.split("T", 1)[0] return datetime.fromisoformat(dt_str).date() except Exception: logger.warning("Impossible de parser dataSituacionAdministrativa : %s", dt_str) return None def etat_to_int(etat: str) -> int: """ Convertit l'état texte ('Pechada' / 'Aberta' / ...) en 0/1. - 1 = fermé (Pechada) - 0 = ouvert (tout le reste) """ if not etat: return 0 etat_norm = etat.strip().lower() if etat_norm.startswith("pech"): return 1 return 0 def toxines_to_str(toxicidade) -> str | None: """ Transforme la liste 'toxicidade' en string pour la base. Exemples : [] -> None ["Lipofílicas"] -> "Lipofílicas" ["Lipofílicas", "ASP"] -> "Lipofílicas,ASP" """ if not toxicidade: return None if isinstance(toxicidade, list): items = [str(x).strip() for x in toxicidade if str(x).strip()] return ",".join(items) if items else None # Si c'est déjà une chaîne s = str(toxicidade).strip() return s or None # --------------------------------------------------------------------------- # Récupération des sites (mapping nom_poligono -> lignes) # --------------------------------------------------------------------------- def fetch_sites_mapping(conn): """ Récupère les sites depuis analyses_sites, en s'appuyant sur nom_poligono. Retourne un dict : { 'Sada 1': [ {id: ..., nom: ..., nom_poligono: ...}, ... ], ... } pour gérer les duplications de poligono (ex: Sada 2 dans 2 zones). """ sql = """ SELECT id, nom, nom_poligono FROM analyses_sites WHERE nom_poligono IS NOT NULL AND nom_poligono <> '' """ mapping = {} with conn.cursor(dictionary=True) as cur: cur.execute(sql) rows = cur.fetchall() for row in rows: key = (row["nom_poligono"] or "").strip() if not key: continue mapping.setdefault(key, []).append(row) return mapping # --------------------------------------------------------------------------- # transformation de "pechada" -> 1, autre -> 0 # --------------------------------------------------------------------------- def etat_to_tinyint(etat: str) -> int: """ Convertit l'état texte de l'API en tinyint : - 'Pechada' -> 1 (fermé) - tout le reste -> 0 (ouvert) """ if not etat: return 0 etat_norm = etat.strip().lower() if etat_norm.startswith("pech"): return 1 return 0 # --------------------------------------------------------------------------- # Mise à jour de l'état courant dans analyses_sites # --------------------------------------------------------------------------- def update_site_current_state( conn, id_site: int, plan: str | None, toxines: str | None, etat_administratif: int, date_changement_etat, ): """ Met à jour les colonnes d'état dans analyses_sites pour un site donné. Paramètres ---------- conn : mysql.connector.connection_cext.CMySQLConnection Connexion MySQL ouverte. id_site : int Valeur de analyses_sites.id (clé primaire). plan : str | None Code du plan (A, B1, B2, C1, ...), ou None. toxines : str | None Liste de toxines sous forme de chaîne (ex: "Lipofílicas;ASP"), ou None. etat_administratif : int 0 = ouvert, 1 = fermé. date_changement_etat : date Date de changement d'état. """ # On force en 0 / 1 pour éviter toute surprise etat_val = 1 if etat_administratif else 0 sql = """ UPDATE analyses_sites SET plan = %(plan)s, toxines = %(toxines)s, etat_administratif = %(etat_administratif)s, date_changement_etat = %(date_changement_etat)s WHERE id = %(id_site)s """ params = { "plan": plan, "toxines": toxines, "etat_administratif": etat_val, "date_changement_etat": date_changement_etat, "id_site": id_site, } try: with conn.cursor() as cur: cur.execute(sql, params) # commit géré à l'extérieur (dans main), comme tu le fais déjà except Exception as e: # log si logger global dispo try: logger.error( "Erreur lors de la mise à jour de l'état pour le site id=%s : %s", id_site, e, ) except NameError: # au cas où pas de logger global print(f"[ERREUR] update_site_current_state id={id_site} : {e}") raise # --------------------------------------------------------------------------- # Insertion dans l'historique, si nouveau changement # --------------------------------------------------------------------------- def insert_history_if_new( conn, id_site: int, date_document, heure_document, plan: str | None, toxines: str | None, etat_administratif: int, date_changement_etat, ): """ Vérifie si une ligne d'historique existe déjà pour (id_site, date_changement_etat). Si non, insère une nouvelle ligne dans analyses_sites_historique. Paramètres ---------- conn : mysql.connector.connection_cext.CMySQLConnection Connexion MySQL ouverte. id_site : int Clé primaire analyses_sites.id. date_document : date Date de l'API (dataActualizacion). heure_document : time | None Heure de l'API (dataActualizacion). plan : str | None Code du plan (A, B1, B2, C1, ...), ou None. toxines : str | None Liste de toxines sous forme de chaîne, ou None. etat_administratif : int 0 = ouvert, 1 = fermé. date_changement_etat : date | None Date de changement d'état renvoyée par l'API. """ if date_changement_etat is None: # Sans date de changement, on ne stocke pas d’historique return # Force 0/1 etat_val = 1 if etat_administratif else 0 check_sql = """ SELECT id_historique FROM analyses_sites_historique WHERE id_site = %(id_site)s AND date_changement_etat = %(date_changement_etat)s LIMIT 1 """ insert_sql = """ INSERT INTO analyses_sites_historique ( id_site, date_document, heure_document, plan, toxines, etat_administratif, date_changement_etat, created_at ) VALUES ( %(id_site)s, %(date_document)s, %(heure_document)s, %(plan)s, %(toxines)s, %(etat_administratif)s, %(date_changement_etat)s, NOW() ) """ params_check = { "id_site": id_site, "date_changement_etat": date_changement_etat, } try: with conn.cursor(dictionary=True) as cur: # On vérifie d'abord si on a déjà cette date pour ce site cur.execute(check_sql, params_check) row = cur.fetchone() if row: # Déjà historisé, on ne duplique pas return params_insert = { "id_site": id_site, "date_document": date_document, "heure_document": heure_document, "plan": plan, "toxines": toxines, "etat_administratif": etat_val, "date_changement_etat": date_changement_etat, } cur.execute(insert_sql, params_insert) except Exception as e: try: logger.error( "Erreur lors de l'insertion historique pour le site id=%s : %s", id_site, e, ) except NameError: print(f"[ERREUR] insert_history_if_new id={id_site} : {e}") raise # --------------------------------------------------------------------------- # Récupération API # --------------------------------------------------------------------------- def fetch_api_data() -> dict: headers = { "User-Agent": "Mozilla/5.0 (compatible; import_situations_api/1.2)" } resp = requests.get(API_URL, headers=headers, timeout=30) resp.raise_for_status() return resp.json() # --------------------------------------------------------------------------- # Traitement principal # --------------------------------------------------------------------------- def main(): start_time = datetime.now() logger.info("Démarrage du script import_situations_api") conn = None try: try: conn = get_db_connection() except Exception: logger.error("Impossible de se connecter à la base, arrêt.") return # 1) Appel API try: data = fetch_api_data() except Exception as e: logger.error("Erreur lors de l'appel à l'API : %s", e) logger.debug("Traceback : %s", traceback.format_exc()) return data_actualizacion = data.get("dataActualizacion") date_document, heure_document = parse_iso_datetime(data_actualizacion) 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 # 2) Mapping analyses_sites.nom_poligono -> lignes mapping = fetch_sites_mapping(conn) logger.info( "Mapping sites chargé : %d noms_poligono distincts", len(mapping), ) updated_count = 0 historised_count = 0 skipped_no_match = 0 # 3) Traitement des zones BI_CV (mytiliculture) for item in estado_zonas: tipo = item.get("tipo", "") # On ne garde que les zones de biotoxines pour les bateas (BI_CV) if tipo != "BI_CV": continue nome = (item.get("nome") or "").strip() if not nome: continue # Correspondance avec nom_poligono sites_for_nome = mapping.get(nome) if not sites_for_nome: skipped_no_match += 1 continue estado_txt = item.get("estado", "") # "Pechada" / "Aberta" / ... estado_detalle = item.get("estadoDetalle") # plan : A, B1, C1, ... 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) # Application à tous les sites correspondant à ce poligono (doublons possibles) 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, ) try: update_site_current_state( conn, 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, 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 e: logger.error( "Erreur lors de la mise à jour/historisation du site id=%s : %s", id_site, e, ) logger.debug("Traceback : %s", traceback.format_exc()) # Commit global try: conn.commit() except Exception as e: logger.error("Erreur lors du commit : %s", e) logger.debug("Traceback : %s", traceback.format_exc()) 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 e: logger.error("Erreur inattendue dans import_situations_api : %s", e) 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, ) if __name__ == "__main__": main()