""" Moteur de résultats – Version TP + DETAIL brut (patch final) ----------------------------------------------------------- Étapes couvertes : 1) Lire les courses actives (reader_config). 2) Pour chaque course active : - Charger la configuration TP / contrôleurs (seulement pour ce con_id !) - Lire resultv2__lecture.last_processed_read_id - Lire les nouvelles lectures dans reader_lectures - Associer chaque lecture à un coureur (cla_list_tag) - Associer chaque lecture à un TP (via chrinfo_location → config_detail) - Écrire une ligne brute dans resultv2__detail avec con_id, dossard, tp_id, tp_name, chrinfo_location, priorité, controller_name, etc. """ from typing import Optional, Dict, List from mysql.connector import Error from engine.db import get_connection BATCH_SIZE = 1000 # --------------------------------------------------------- # 1) COURSES ACTIVES # --------------------------------------------------------- def get_active_con_ids(cur) -> List[int]: """Retourne tous les con_id actifs.""" sql = "SELECT con_id FROM reader_config WHERE rea_active = 1" cur.execute(sql) rows = cur.fetchall() return [int(r["con_id"]) for r in rows] # --------------------------------------------------------- # 2) CHARGER LA CONFIG TP / CONTRÔLEURS (PAR COURSE) # --------------------------------------------------------- def load_tp_config_for_con(cur, con_id: int) -> Dict[str, List[Dict]]: """ Charge TOUTE la configuration TP pour UNE course (con_id). Retourne un dict : clé = chrinfo_location (ex: 'PRO5655') valeur= liste de dicts { con_id, tp_id, tp_name, priority, controller_name } """ sql = """ SELECT tp.tp_id, tp.con_id, tp.tp_name, d.det_value1 AS chr_location, d.det_value2 AS priority, d.det_value3 AS controller_name FROM reader_config_timingpoint tp JOIN reader_config_detail d ON d.det_type = 1 AND d.det_ref1_id = tp.tp_id AND d.det_clef = 'clef_controleur' WHERE tp.con_id = %s """ cur.execute(sql, (con_id,)) rows = cur.fetchall() tp_map: Dict[str, List[Dict]] = {} for row in rows: loc = row["chr_location"] if loc not in tp_map: tp_map[loc] = [] try: prio = int(row["priority"]) if row["priority"] is not None else 1 except ValueError: prio = 1 tp_map[loc].append( { "con_id": int(row["con_id"]), "tp_id": int(row["tp_id"]), "tp_name": row["tp_name"], "priority": prio, "controller_name": row["controller_name"], } ) print(f"[DEBUG] TP config chargée pour CON_ID={con_id}:") if not tp_map: print(" (aucune ligne trouvée pour ce con_id)") else: for loc, items in tp_map.items(): parts = [ f"tp_id={tp['tp_id']} tp_name={tp['tp_name']} " f"prio={tp['priority']} name={tp['controller_name']}" for tp in items ] print(f" location={loc} -> " + " | ".join(parts)) return tp_map # --------------------------------------------------------- # 3) RÉSOUDRE LE TP POUR UNE LECTURE # --------------------------------------------------------- def resolve_tp_for_lecture(tp_map: Dict[str, List[Dict]], con_id: int, chr_location: str) -> Optional[Dict]: """Trouve le TP applicable pour chrinfo_location avec ce con_id.""" if chr_location not in tp_map: return None candidates = [tp for tp in tp_map[chr_location] if tp["con_id"] == con_id] if not candidates: return None return sorted(candidates, key=lambda t: t["priority"])[0] # --------------------------------------------------------- # 4) LECTURE DE L'ÉTAT MOTEUR # --------------------------------------------------------- def get_last_processed_read_id(cur, engine_table: str) -> Optional[int]: sql = f"SELECT last_processed_read_id FROM `{engine_table}` WHERE id = 1" cur.execute(sql) row = cur.fetchone() return int(row["last_processed_read_id"]) if row else None # --------------------------------------------------------- # 5) LECTURES BRUTES # --------------------------------------------------------- def get_new_lectures(cur, last_processed: int): sql = """ SELECT rcourse_id, chrinfo_tag, chrinfo_time, chrinfo_location FROM reader_lectures WHERE rcourse_id > %s ORDER BY rcourse_id ASC LIMIT %s """ cur.execute(sql, (last_processed, BATCH_SIZE)) return cur.fetchall() # --------------------------------------------------------- # 6) MATCH TAG → COUREUR # --------------------------------------------------------- def get_classement_by_tag(cur, classement_table: str, tag: str) -> Optional[Dict]: pattern = f"%*{tag}*%" sql = f""" SELECT cla_info_dossard, con_id, cla_list_tag FROM `{classement_table}` WHERE cla_list_tag LIKE %s LIMIT 1 """ cur.execute(sql, (pattern,)) return cur.fetchone() # --------------------------------------------------------- # 7) INSERT DANS DETAIL # --------------------------------------------------------- def insert_detail_record( cur, cnx, con_id: int, dossard: str, lecture: Dict, tp_id: int, tp_name: str, controller_location: str, priority: int, controller_name: str, ): detail_table = f"resultv2_{con_id}_detail" sql = f""" INSERT INTO `{detail_table}` (con_id, tp_id, tp_name, cla_info_dossard, rcourse_id, timestamp_read, raw_tag, reader_location, controller_priority, controller_name, source) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ data = ( con_id, tp_id, tp_name, dossard, lecture["rcourse_id"], str(lecture["chrinfo_time"]), lecture["chrinfo_tag"], controller_location, priority, controller_name, "LIVE", ) cur.execute(sql, data) cnx.commit() # --------------------------------------------------------- # 8) METTRE À JOUR LE POINTEUR # --------------------------------------------------------- def update_last_processed_read_id(cur, cnx, engine_table: str, new_value: int): sql = f""" UPDATE `{engine_table}` SET last_processed_read_id = %s, updated_at = NOW() WHERE id = 1 """ cur.execute(sql, (new_value,)) cnx.commit() # --------------------------------------------------------- # 9) TRAITER UNE COURSE # --------------------------------------------------------- def process_course(cur, cnx, con_id: int): engine_table = f"resultv2_{con_id}_lecture" classement_table = f"resultv2_{con_id}_classement" # CHARGER la config TP / contrôleurs **PAR course** tp_map = load_tp_config_for_con(cur, con_id) last_processed = get_last_processed_read_id(cur, engine_table) if last_processed is None: print(f"[ERREUR] Pas de ligne id=1 dans {engine_table}.") return print(f"[INFO] CON_ID={con_id} last_processed_read_id={last_processed}") lectures = get_new_lectures(cur, last_processed) if not lectures: print(f"[INFO] CON_ID={con_id} : aucune nouvelle lecture.") return max_rcourse_id = last_processed for lecture in lectures: rcourse_id = int(lecture["rcourse_id"]) tag = str(lecture["chrinfo_tag"]) chr_location = str(lecture["chrinfo_location"]) max_rcourse_id = max(max_rcourse_id, rcourse_id) # TAG → COUREUR coureur = get_classement_by_tag(cur, classement_table, tag) if not coureur: print(f"[WARN] CON_ID={con_id} rcourse_id={rcourse_id}: tag {tag} introuvable") continue dossard = coureur["cla_info_dossard"] # LOCATION → TP tp_info = resolve_tp_for_lecture(tp_map, con_id, chr_location) if not tp_info: print( f"[WARN] CON_ID={con_id} rcourse_id={rcourse_id}: " f"location {chr_location} introuvable pour cette course" ) continue tp_id = tp_info["tp_id"] tp_name = tp_info["tp_name"] priority = tp_info["priority"] controller_name = tp_info["controller_name"] print( f"[DEBUG] CON_ID={con_id} rcourse_id={rcourse_id} " f"loc={chr_location} -> tp_id={tp_id} tp_name={tp_name} " f"prio={priority} name={controller_name}" ) # INSERT DETAIL insert_detail_record( cur, cnx, con_id, dossard, lecture, tp_id, tp_name, chr_location, priority, controller_name, ) print( f"[OK] CON_ID={con_id} rcourse_id={rcourse_id} " f"tag={tag} → dossard={dossard}, tp={tp_name} (id={tp_id})" ) update_last_processed_read_id(cur, cnx, engine_table, max_rcourse_id) print(f"[OK] CON_ID={con_id} last_processed_read_id → {max_rcourse_id}") # --------------------------------------------------------- # 10) BOUCLE PRINCIPALE # --------------------------------------------------------- def run_once(): try: cnx = get_connection() cur = cnx.cursor(dictionary=True) con_ids = get_active_con_ids(cur) if not con_ids: print("[INFO] Aucune course active.") return for con_id in con_ids: process_course(cur, cnx, con_id) except Error as e: print(f"[ERREUR MYSQL] {e}") finally: try: cur.close() cnx.close() except: pass if __name__ == "__main__": run_once()