#!/usr/bin/env python3 """Hermes Football Link Manager V2: auditoría ESPN/EPG/M3U, sin cambios en producción. Python 3.8. No descarga ni reproduce streams. No imprime URLs ni credenciales. """ import argparse import json import re import sqlite3 import unicodedata from datetime import datetime, timezone, timedelta from pathlib import Path from urllib.parse import urlsplit STOP = {'fc','cf','sc','afc','club','de','the','04','vs','contra','sv','fk'} def norm(s): s = ''.join(c for c in unicodedata.normalize('NFKD',str(s or '').lower()) if not unicodedata.combining(c)) return re.sub(r'[^a-z0-9]+',' ',s).strip() def tokens(s): return set(norm(s).split()) - STOP def epg_date(s): return datetime.strptime(s, '%Y%m%d%H%M%S %z').astimezone(timezone.utc) def load_playlist(path): """Retorna nombres y tvg-id; intencionalmente descarta URL de reproducción.""" if not path: return [] result=[] ext='' with open(path,'r',encoding='utf-8-sig',errors='replace') as fh: for line in fh: line=line.strip() if line.startswith('#EXTINF'): ext=line elif line and not line.startswith('#') and ext: name=ext.rsplit(',',1)[-1].strip() tvg=re.search(r'tvg-id="([^"]+)"',ext,re.I) group=re.search(r'group-title="([^"]+)"',ext,re.I) if 'vix' in norm(name) and 'deportes' in norm(name) and ('sling' in norm(name) or 'sling' in norm(group.group(1) if group else '')): result.append({'name':name,'tvg_id':tvg.group(1) if tvg else ''}) ext='' return result def channel_num(text): t=norm(text) m=re.search(r'vix\s*(?:premium\s*)?deportes\s*(\d+)',t) if m: return ('premium' if 'premium' in t else 'normal',int(m.group(1))) return None def epg_num(canal): m=re.match(r'vix(premium)?deportes(\d+)\.us$',str(canal).lower()) return ('premium' if m.group(1) else 'normal',int(m.group(2))) if m else None def main(): p=argparse.ArgumentParser(description='Cruce de partidos EPG y catálogo Sling, sin escritura en DB') p.add_argument('--fecha',required=True,help='YYYY-MM-DD, fecha local de tarjetas') p.add_argument('--football',default='/var/lib/hermes-nexus/football.db') p.add_argument('--epg',default='/var/lib/hermes-nexus/epg.db') p.add_argument('--playlist',help='Ruta LOCAL a M3U descargado previamente; opcional') p.add_argument('--salida',help='Ruta del informe JSON; no contiene URLs de reproducción') args=p.parse_args() day=datetime.strptime(args.fecha,'%Y-%m-%d').date() football=sqlite3.connect('file:'+args.football+'?mode=ro',uri=True) epg=sqlite3.connect('file:'+args.epg+'?mode=ro',uri=True) football.row_factory=sqlite3.Row epg.row_factory=sqlite3.Row events=[dict(x) for x in football.execute('SELECT * FROM events WHERE local_day=?',(args.fecha,))] # El día local de Mexicali puede abarcar dos fechas UTC: ampliar ventana de consulta. start=(datetime.combine(day,datetime.min.time())-timedelta(days=1)).strftime('%Y%m%d%H%M%S') end=(datetime.combine(day,datetime.min.time())+timedelta(days=2)).strftime('%Y%m%d%H%M%S') programs=[dict(x) for x in epg.execute('''SELECT canal,titulo,inicio,fin FROM epg_programas WHERE fuente='sling' AND (canal LIKE 'vixpremiumdeportes%' OR canal LIKE 'vixdeportes%') AND inicio < ? AND fin > ?''',(end,start))] football.close();epg.close() channels=load_playlist(args.playlist) output=[] for event in events: home=tokens(event.get('home')); away=tokens(event.get('away')) try: event_dt=datetime.fromisoformat(event['starts_utc'].replace('Z','+00:00')).astimezone(timezone.utc) except (TypeError,ValueError,AttributeError): event_dt=None found=[] for pr in programs: words=tokens(pr['titulo']) if not home or not away or not all(len(team & words)>=max(1,len(team)-1) for team in (home,away)): continue try: delta=round((epg_date(pr['inicio'])-event_dt).total_seconds()/60) if event_dt else None except (ValueError,TypeError): delta=None if delta is None or abs(delta)>90: continue num=epg_num(pr['canal']) catalog=[x for x in channels if channel_num(x['name'])==num] if num else [] # Duplicados EPG con el mismo canal son un solo candidato. found.append({'epg_canal':pr['canal'],'epg_titulo':pr['titulo'], 'epg_inicio':pr['inicio'],'diferencia_minutos':delta, 'catalogo_coincidencias':len(catalog) if args.playlist else None, 'catalogo_nombre':catalog[0]['name'] if len(catalog)==1 else None, 'estado':'HORARIO_REVISAR' if abs(delta)>20 else ('CATALOGO_UNICO' if len(catalog)==1 else ('CATALOGO_AMBIGUO' if len(catalog)>1 else 'SIN_CANAL_M3U')) if args.playlist else 'EPG_PROPUESTO'}) unique={} for x in found: k=x['epg_canal'] if k not in unique or abs(x['diferencia_minutos'])1 else 'SIN_COINCIDENCIA' if not matches else 'PENDIENTE') output.append({'id':event.get('id'),'partido':str(event.get('away'))+' contra '+str(event.get('home')), 'inicio_espn':event.get('starts_utc'),'estado':state,'coincidencias':matches}) report={'version':'V2','modo':'SOLO_LECTURA','fecha':args.fecha,'total_eventos':len(events), 'm3u_local_proporcionado':bool(args.playlist),'canales_sling_m3u':len(channels), 'nota':'No verifica reproducción ni derechos; ninguna URL de transmisión se guarda o publica.', 'resultados':output} rendered=json.dumps(report,ensure_ascii=False,indent=2) if args.salida: dest=Path(args.salida).expanduser() if dest.exists():raise SystemExit('La salida ya existe: no se sobrescribe') dest.write_text(rendered+'\n',encoding='utf-8') dest.chmod(0o600) print('Informe guardado:',dest,'| eventos:',len(events),'| propuestas:',sum(x['estado']=='PROPUESTA_PARA_REVISION' for x in output)) else:print(rendered) if __name__=='__main__':main()