#!/usr/bin/env python3 """Hermes LMP V3 DRY RUN: no network, no O11 calls, no production changes. Python 3.8+.""" import argparse, json, re, unicodedata from datetime import datetime, timedelta from pathlib import Path import pytz TZ = pytz.timezone('America/Tijuana') ALIASES = {'aguilas':['aguilas','mexicali'], 'naranjeros':['naranjeros','hermosillo'], 'yaquis':['yaquis','obregon'], 'mayos':['mayos','navojoa'], 'caneros':['caneros','mochis'], 'algodoneros':['algodoneros','guasave'], 'tomateros':['tomateros','culiacan'], 'venados':['venados','mazatlan'], 'charros':['charros','jalisco'], 'jaguares':['jaguares','nayarit','tepic']} RETRY_MINUTES = [0, 2, 5, 10, 18] def norm(s): s = ''.join(c for c in unicodedata.normalize('NFD', str(s).lower()) if unicodedata.category(c) != 'Mn') return re.sub('[^a-z0-9]', '', s) def identify(s): n = norm(s) found = [k for k, names in ALIASES.items() if any(norm(x) in n for x in names)] return found[0] if len(found) == 1 else None def dt(s): if not s: return None d = datetime.fromisoformat(str(s).replace('Z', '+00:00')) return TZ.localize(d) if d.tzinfo is None else d.astimezone(TZ) def main(): p = argparse.ArgumentParser(description='Simulacion LMP V3: cero solicitudes de red') p.add_argument('epg_json') p.add_argument('--playlist', help='Snapshot local playlist.m3u, previamente guardado (opcional)') p.add_argument('--state', default='hermes_lmp_v3_state.json') p.add_argument('--now', help='Fecha ISO para simulacion') p.add_argument('--assume-final-2', action='store_true', help='SOLO PRUEBAS: asumir que EPG 2 es final') a = p.parse_args() now = dt(a.now) if a.now else datetime.now(TZ) epg = json.loads(Path(a.epg_json).read_text(encoding='utf-8')) response = epg.get('response', {}) games = [] for section in ('games', 'prevGames', 'nextGames'): games.extend(g for g in (response.get(section) or []) if isinstance(g, dict)) path = Path(a.state) state = json.loads(path.read_text(encoding='utf-8')) if path.exists() else {'games':{}, 'provider':{}} if not isinstance(state, dict): raise ValueError('Estado invalido') state.setdefault('games', {}); state.setdefault('provider', {}) if not isinstance(state['games'], dict) or not isinstance(state['provider'], dict): raise ValueError('Estado invalido') present = None if a.playlist: content = Path(a.playlist).read_text(encoding='utf-8') present = set() # Snapshot administrativo: comparar solo nombres EXTINF, nunca solicitar streams. for line in content.splitlines(): if line.startswith('#EXTINF:'): label = norm(line.split(',', 1)[-1]) for slug, aliases in ALIASES.items(): if any(norm(x) in label for x in aliases): present.add(slug) unique = {} for g in games: away, home = identify(g.get('awayTeamShort', '')), identify(g.get('homeTeamShort', '')) start = dt(g.get('localFullDate')) if not away or not home or away == home or not start: continue key = start.isoformat()+'|'+min(away,home)+'|'+max(away,home) unique[key] = (g,away,home,start) print('=== HERMES LMP V3: SOLO SIMULACION ===') print('HORA:', now.isoformat(), '| PARTIDOS:', len(unique), '| PLAYLIST:', 'SIN SNAPSHOT' if present is None else 'SNAPSHOT LOCAL') due = [] for key,(g,away,home,start) in sorted(unique.items(), key=lambda item:item[1][3]): entry = state['games'].setdefault(key,{}) code = g.get('gameStatus') if code == 1: entry.pop('first_final_seen', None) decision = 'EN VIVO: mantener acceso' elif code == 2 and a.assume_final_2: entry.setdefault('first_final_seen', now.isoformat()) first = dt(entry['first_final_seen']) decision = 'SIMULAR CIERRE' if now >= first+timedelta(minutes=30) else 'MARGEN POSTPARTIDO' elif code == 2: decision = 'ESTADO 2 SIN VALIDAR: NO CERRAR' elif code not in (0,1,2): decision = 'ESTADO DESCONOCIDO: NO CERRAR' elif now < start-timedelta(minutes=10): decision = 'PREVIO: no abrir' elif now > start+timedelta(minutes=30): decision = 'FUERA DE VENTANA: no reintentar' else: if present is None: decision = 'PENDIENTE VERIFICACION: falta snapshot administrativo' elif away in present and home in present: decision = 'CANALES PRESENTES: sin actualizacion' else: decision = 'FALTANTES: '+','.join(x for x in (away,home) if x not in present) due.append((key, start-timedelta(minutes=10), start+timedelta(minutes=30))) print('{} vs {} | {} | EPG={} | {}'.format(away,home,start.isoformat(),code,decision)) # Un solo estado para proveedor; cada llamada simulada se registra y evita duplicados entre ejecuciones. if due: provider = state['provider'] windows = provider.setdefault('windows', {}) # Agrupar preparaciones cercanas (maximo 5 minutos). grupos = [] for key, preparar, limite in sorted(due, key=lambda x: x[1]): if not grupos or preparar - grupos[-1]['anchor'] > timedelta(minutes=5): grupos.append({ 'anchor': preparar, 'latest': limite, 'games': [key] }) else: grupos[-1]['games'].append(key) grupos[-1]['latest'] = max(grupos[-1]['latest'], limite) acciones = 0 for grupo in grupos: anchor = grupo['anchor'] latest = grupo['latest'] window_id = anchor.isoformat() window = windows.setdefault(window_id, { 'attempts': [], 'games': grupo['games'] }) attempts = window['attempts'] next_index = len(attempts) if next_index >= len(RETRY_MINUTES): print('VENTANA:', window_id, 'LIMITE ALCANZADO') elif now < anchor + timedelta(minutes=RETRY_MINUTES[next_index]): print('VENTANA:', window_id, 'ESPERANDO') elif now > latest: print('VENTANA:', window_id, 'FUERA DE PLAZO') else: attempts.append(now.isoformat()) acciones += 1 print( 'O11 SIMULADO: VENTANA', window_id, 'INTENTO', len(attempts), 'DE', len(RETRY_MINUTES) ) print('TOTAL ACTUALIZACIONES SIMULADAS:', acciones) else: print('O11 SIMULADO: NINGUNA ACTUALIZACION NECESARIA') # Atomic write for local simulation state only. tmp = path.with_name(path.name+'.tmp') tmp.write_text(json.dumps(state,indent=2,ensure_ascii=False),encoding='utf-8') tmp.replace(path) print('ESTADO LOCAL:',path,'| O11: CERO PETICIONES | PRODUCCION: SIN CAMBIOS') if __name__ == '__main__': main()