#!/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 base64 import getpass import os import sys import urllib.request import urllib.error 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 download_catalog(url, username, password, limit=25*1024*1024): """Consulta O11 y conserva solo metadatos EXTINF; nunca guarda URLs de streams.""" if not url.startswith(('http://','https://')): raise ValueError('La URL del catalogo debe ser HTTP(S)') auth=base64.b64encode((username+':'+password).encode()).decode('ascii') req=urllib.request.Request(url,headers={ 'Authorization':'Basic '+auth, 'Accept':'*/*', 'User-Agent':'Mozilla/5.0 (Hermes-Nexus; catalog-audit)' }) with urllib.request.urlopen(req,timeout=45) as response: payload=response.read(limit+1) if len(payload)>limit: raise ValueError('Playlist excede el limite de 25 MB') result=[] for line in payload.decode('utf-8-sig',errors='replace').splitlines(): if not line.startswith('#EXTINF'): continue name=line.rsplit(',',1)[-1].strip() tvg=re.search(r'tvg-id="([^"]+)"',line,re.I) group=re.search(r'group-title="([^"]+)"',line,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 ''}) 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='M3U local opcional; si se omite, descarga O11') p.add_argument('--o11-url',default='http://15.204.47.5:1234/playlist.m3u',help='URL del catalogo O11') p.add_argument('--o11-user',default=os.environ.get('HERMES_O11_USER','admin'),help='Usuario O11') p.add_argument('--sin-o11',action='store_true',help='No descargar catalogo') 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() catalog_used=not args.sin_o11 if args.sin_o11: channels=[] elif args.playlist: channels=load_playlist(args.playlist) else: password=os.environ.get('HERMES_O11_PASSWORD') if not password: if not sys.stdin.isatty(): raise SystemExit('Falta HERMES_O11_PASSWORD (o ejecuta en terminal para ingresar clave)') password=getpass.getpass('Clave O11 Sling (no se guarda): ') try: channels=download_catalog(args.o11_url,args.o11_user,password) except (urllib.error.URLError,ValueError,OSError) as exc: raise SystemExit('No se pudo descargar el catalogo O11: '+type(exc).__name__) finally: password=None 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 catalog_used 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 catalog_used 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.1','modo':'SOLO_LECTURA','fecha':args.fecha,'total_eventos':len(events), 'catalogo_consultado':catalog_used,'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()