"""Página de Directos — partidos de hoy de los jugadores del CSV. Muestra los partidos del día agrupados por estado: • Programados (notstarted) • En vivo (inprogress) — con stats y auto-refresh cada 60s • Finalizados (finished) Diseñada para no saturar la API: sólo activa el auto-refresh si hay al menos un partido en vivo. Si únicamente hay programados o finalizados, no hay refresh automático. """ import time from datetime import datetime, date import streamlit as st from lib import _get, SOFASCORE_BASE, color_rating, PHOTO_PLACEHOLDER st.title("🔴 Partidos de hoy") st.caption("Partidos del día de los jugadores del CSV — refresco automático cada minuto cuando hay directos") df = st.session_state.get("df") if df is None or df.empty: st.warning("Aún no se han cargado los jugadores.") st.stop() # ── Helpers ──────────────────────────────────────────────────────────────── def _estado_partido(ev): """Devuelve uno de: 'notstarted' | 'inprogress' | 'finished' | 'otro'.""" t = (ev.get("status") or {}).get("type", "otro") return t if t in ("notstarted", "inprogress", "finished") else "otro" def _minuto_actual(ev): """Calcula el minuto aproximado del partido a partir de time.currentPeriodStartTimestamp.""" t = ev.get("time", {}) or {} inicio = t.get("currentPeriodStartTimestamp") initial = t.get("initial", 0) if not inicio: return None seg_transcurridos = max(0, time.time() - inicio) + initial return int(seg_transcurridos // 60) def _hora_kickoff(ev): ts = ev.get("startTimestamp") if not ts: return "—" return datetime.fromtimestamp(ts).strftime("%H:%M") def _info_jugador_en_partido(event_id, player_id): """Combina lineups + incidents para conocer el estado del jugador y sus acciones. Devuelve { "estado": "campo" | "banquillo" | "sustituido" | "no_convocado", "acciones": [{"tipo": str, "minuto": int}, ...] # orden cronológico } """ estado = "no_convocado" acciones = [] # Lineups → titular o banquillo try: lineups = _get(f"{SOFASCORE_BASE}/event/{event_id}/lineups") or {} for side in ("home", "away"): for p in (lineups.get(side, {}).get("players") or []): if (p.get("player") or {}).get("id") == player_id: estado = "banquillo" if p.get("substitute") else "campo" break except Exception: pass # Incidents → eventos del partido try: inc_data = _get(f"{SOFASCORE_BASE}/event/{event_id}/incidents") or {} incidents = inc_data.get("incidents", []) or [] except Exception: incidents = [] for inc in incidents: tipo = inc.get("incidentType") t = inc.get("time") if tipo == "substitution": if (inc.get("playerOut") or {}).get("id") == player_id: estado = "sustituido" acciones.append({"tipo": "subOut", "minuto": t}) if (inc.get("playerIn") or {}).get("id") == player_id: estado = "campo" # entró desde el banquillo acciones.append({"tipo": "subIn", "minuto": t}) elif tipo == "goal": if (inc.get("player") or {}).get("id") == player_id: clase = inc.get("incidentClass", "regular") if clase == "ownGoal": acciones.append({"tipo": "ownGoal", "minuto": t}) elif clase == "penalty": acciones.append({"tipo": "penalty", "minuto": t}) else: acciones.append({"tipo": "gol", "minuto": t}) assist = inc.get("assist1") or {} if assist.get("id") == player_id: acciones.append({"tipo": "asistencia", "minuto": t}) elif tipo == "card": if (inc.get("player") or {}).get("id") == player_id and not inc.get("rescinded"): clase = inc.get("incidentClass", "yellow") acciones.append({"tipo": f"card_{clase}", "minuto": t}) # Ordenar por minuto (None va al final) acciones.sort(key=lambda a: (a.get("minuto") is None, a.get("minuto") or 0)) return {"estado": estado, "acciones": acciones} _BADGE_ESTADO = { "campo": ("🟢", "EN EL CAMPO", "#27ae60", "white"), "banquillo": ("🪑", "EN BANQUILLO", "#7f8c8d", "white"), "sustituido": ("🔄", "SUSTITUIDO", "#c0392b", "white"), "no_convocado": ("⚪", "NO CONVOCADO", "#bdc3c7", "#333"), } def _badge_estado_jugador(estado): icono, txt, bg, fg = _BADGE_ESTADO.get(estado, _BADGE_ESTADO["no_convocado"]) return ( f'{icono} {txt}' ) _ICONOS_ACCION = { "gol": ("⚽", "Gol", "#27ae60"), "penalty": ("⚽p", "Gol de penalti", "#27ae60"), "ownGoal": ("🥅", "Gol en propia", "#c0392b"), "asistencia": ("🅰️", "Asistencia", "#3498db"), "subIn": ("🔼", "Entró", "#27ae60"), "subOut": ("🔽", "Salió", "#7f8c8d"), "card_yellow": ("🟨", "Amarilla", "#f39c12"), "card_yellowRed": ("🟨🟥", "Doble amarilla", "#c0392b"), "card_red": ("🟥", "Roja", "#c0392b"), } def _badge_accion(accion): tipo = accion.get("tipo") minuto = accion.get("minuto") icono, etiqueta, color = _ICONOS_ACCION.get(tipo, ("•", tipo or "?", "#7f8c8d")) min_txt = f"{minuto}'" if minuto is not None else "" return ( f'' f'{icono} {min_txt}' ) def _resaltar_palos_y_chances_grandes(stats): """Genera badges adicionales a partir de stats que no están en incidents.""" extras = [] if stats.get("hitWoodwork"): extras.append(( f'🪵 Palo ×{int(stats["hitWoodwork"])}' )) if stats.get("bigChanceMissed"): extras.append(( f'🎯💨 Ocas. fallada ×{int(stats["bigChanceMissed"])}' )) return extras def _buscar_partidos_hoy(df_jugadores): """Para cada jugador, busca su partido de hoy (cualquier estado). SofaScore expone dos endpoints distintos: events/last (pasados / en curso) y team/{tid}/events/next (próximos). Combinamos ambos y desduplicamos por event id. Sin cache para frescura. """ hoy = date.today() resultados = [] # Cache local en este render para no repetir llamadas si hay varios jugadores del mismo equipo cache_team_next = {} for _, j in df_jugadores.iterrows(): eventos = [] # Pasados / en curso try: data = _get(f"{SOFASCORE_BASE}/player/{j['id']}/events/last/0") if isinstance(data, dict): eventos.extend(data.get("events", []) or []) except Exception: pass # Próximos del equipo equipo_id = j.get("equipo_id") if equipo_id is not None: try: tid = int(equipo_id) if tid not in cache_team_next: data_next = _get(f"{SOFASCORE_BASE}/team/{tid}/events/next/0") cache_team_next[tid] = (data_next or {}).get("events", []) or [] eventos.extend(cache_team_next[tid]) except (TypeError, ValueError): pass # Desduplicar por event id, quedándonos con la versión más actualizada # (la de events/last si el partido ya está en juego trae el status correcto) vistos = {} for ev in eventos: eid = ev.get("id") if not eid: continue # Si ya está, mantener el que tenga score (events/last suele estar más fresco) if eid in vistos: if (ev.get("homeScore") or {}).get("current") is not None and (vistos[eid].get("homeScore") or {}).get("current") is None: vistos[eid] = ev else: vistos[eid] = ev # Buscar el de hoy for ev in vistos.values(): ts = ev.get("startTimestamp", 0) if ts and datetime.fromtimestamp(ts).date() == hoy: resultados.append({"jugador": j.to_dict(), "evento": ev}) break # un partido por jugador return resultados def _render_marcador(ev, j, grande=False): """HTML del marcador con el equipo del jugador en negrita.""" home = ev.get("homeTeam", {}) or {} away = ev.get("awayTeam", {}) or {} hs = ev.get("homeScore", {}).get("current") as_ = ev.get("awayScore", {}).get("current") es_local = home.get("id") == j.get("equipo_id") score = "—" if hs is None or as_ is None else f"{hs}–{as_}" tam = "24px" if grande else "16px" if es_local: return ( f'{home.get("name", "—")}' f' {score} ' f'{away.get("name", "—")}' ) return ( f'{home.get("name", "—")}' f' {score} ' f'{away.get("name", "—")}' ) def _render_stats(ev, j, info): """Pinta las acciones destacadas y stats del jugador en el partido (sin cache). `info` es el dict devuelto por `_info_jugador_en_partido` (estado + acciones). """ try: data = _get(f"{SOFASCORE_BASE}/event/{ev['id']}/player/{j['id']}/statistics") stats = (data or {}).get("statistics") or {} except Exception: stats = {} # Acciones destacadas (badges con iconos) acciones_html = "".join(_badge_accion(a) for a in info.get("acciones", [])) acciones_html += "".join(_resaltar_palos_y_chances_grandes(stats)) if acciones_html: st.markdown( f'
{acciones_html}
', unsafe_allow_html=True, ) if not stats: if info.get("estado") == "banquillo": st.caption("🪑 Está en el banquillo, todavía no ha entrado.") else: st.caption("Aún sin estadísticas disponibles.") return rating = stats.get("rating") r_bg, r_fg = color_rating(rating) if rating else ("#cccccc", "#666") col_r, col_min, col_g, col_a, col_t, col_p = st.columns(6) with col_r: rating_html = f'{rating:.1f}' if rating else "—" st.markdown( f'
' f'
Rating
' f'
' f'{rating_html}
', unsafe_allow_html=True, ) col_min.metric("Minutos", stats.get("minutesPlayed", "—")) col_g.metric("⚽ Goles", stats.get("goals", 0)) col_a.metric("👟 Asist.", stats.get("goalAssist", 0)) col_t.metric("Tiros", stats.get("totalShots", 0)) col_p.metric("Pases", stats.get("totalPass", 0)) with st.expander("Más estadísticas"): extras = [ ("onTargetScoringAttempt", "A puerta"), ("keyPass", "Pases clave"), ("accuratePass", "Pases prec."), ("touches", "Toques"), ("duelWon", "Duelos gan."), ("totalTackle", "Entradas"), ("interceptionWon", "Interc."), ("possessionLostCtrl", "Pérdidas"), ] cols_e = st.columns(4) for i, (k, etiqueta) in enumerate(extras): if k in stats: cols_e[i % 4].metric(etiqueta, stats[k]) def _badge_estado(estado, ev): """Badge HTML según el estado del partido.""" if estado == "inprogress": minuto = _minuto_actual(ev) descripcion = (ev.get("status") or {}).get("description", "En juego") if minuto is not None and descripcion not in ("Halftime",): txt = f"⏱️ {minuto}'" else: txt = f"⏱️ {descripcion}" return ( '● EN VIVO' f'{txt}' ) if estado == "notstarted": return ( f'🕒 HOY {_hora_kickoff(ev)}' ) if estado == "finished": return ( '✓ FINALIZADO' ) return "" def _render_partido(p): """Renderiza una tarjeta de partido.""" j = p["jugador"] ev = p["evento"] estado = _estado_partido(ev) foto = j.get("foto") or PHOTO_PLACEHOLDER nombre = j.get("nombre_agencia") or j.get("nombre", "—") # Info de lineup + incidents para los partidos ya empezados info_jug = {"estado": "no_convocado", "acciones": []} if estado in ("inprogress", "finished"): info_jug = _info_jugador_en_partido(ev["id"], int(j["id"])) with st.container(border=True): c_foto, c_info = st.columns([1, 6]) with c_foto: st.markdown( f'', unsafe_allow_html=True, ) with c_info: grande = estado == "inprogress" estado_jug_html = ( _badge_estado_jugador(info_jug["estado"]) if estado in ("inprogress", "finished") else "" ) st.markdown( f'
' f'{nombre}' f'{_badge_estado(estado, ev)}' f'{estado_jug_html}' f'
' f'
{_render_marcador(ev, j, grande=grande)}
' f'
' f'{(ev.get("tournament") or {}).get("name", "—")}
', unsafe_allow_html=True, ) # Stats: sólo si el partido está en vivo o finalizado (los programados aún no tienen) if estado in ("inprogress", "finished"): st.markdown("---") _render_stats(ev, j, info_jug) # ── Cabecera con controles y reloj ───────────────────────────────────────── c_left, c_auto, c_btn = st.columns([5, 2, 1]) with c_btn: if st.button("🔄", help="Refrescar ahora", use_container_width=True): st.rerun() def _refresh_evento(event_id): """Hace fetch al endpoint /event/{eid} y devuelve el dict actualizado o None.""" try: data = _get(f"{SOFASCORE_BASE}/event/{event_id}") if isinstance(data, dict): return data.get("event") or data except Exception: pass return None # ── Cache de session: conserva partidos vistos durante 24h ───────────────── # Estructura: { event_id (str): {"jugador": dict, "evento": dict, "first_seen_ts": float} } # Esto evita que un partido desaparezca cuando SofaScore lo saca de events/next # (porque ya empezó) y events/last no lo trae (porque el jugador no fue convocado). if "directos_cache" not in st.session_state: st.session_state.directos_cache = {} # Limpieza: descartamos entradas con más de 24h o de días distintos al actual ahora_ts = time.time() hoy_str = date.today().isoformat() st.session_state.directos_cache = { k: v for k, v in st.session_state.directos_cache.items() if (ahora_ts - v.get("first_seen_ts", 0)) < 24 * 3600 and v.get("dia") == hoy_str } with st.spinner("Cargando partidos del día..."): partidos_fresh = _buscar_partidos_hoy(df) # Merge: añadimos/actualizamos los frescos en el cache ids_fresh = set() for p in partidos_fresh: eid = str((p.get("evento") or {}).get("id") or "") if not eid: continue ids_fresh.add(eid) if eid in st.session_state.directos_cache: # El evento puede haber cambiado de estado / score → actualizamos st.session_state.directos_cache[eid]["evento"] = p["evento"] st.session_state.directos_cache[eid]["jugador"] = p["jugador"] else: st.session_state.directos_cache[eid] = { "jugador": p["jugador"], "evento": p["evento"], "first_seen_ts": ahora_ts, "dia": hoy_str, } # Para los cacheados que NO vinieron en el fetch fresco, re-pedimos su estado # individual con /event/{eid} (1 petición ligera por evento huérfano) for eid, v in list(st.session_state.directos_cache.items()): if eid in ids_fresh: continue ev_actualizado = _refresh_evento(eid) if ev_actualizado: v["evento"] = ev_actualizado # Lista final = todo lo del cache partidos = [ {"jugador": v["jugador"], "evento": v["evento"]} for v in st.session_state.directos_cache.values() ] partidos.sort(key=lambda p: p["evento"].get("startTimestamp", 0)) en_vivo = [p for p in partidos if _estado_partido(p["evento"]) == "inprogress"] proximos = [p for p in partidos if _estado_partido(p["evento"]) == "notstarted"] finalizados = [p for p in partidos if _estado_partido(p["evento"]) == "finished"] hay_directos = bool(en_vivo) with c_auto: auto_refresh = st.toggle( "Auto-refresh 60s", value=hay_directos, key="directos_auto", help="Sólo se refresca automáticamente si hay partidos en vivo", disabled=not hay_directos, ) with c_left: st.markdown( f'
' f'Última comprobación: {datetime.now().strftime("%H:%M:%S")}' f'{" · 🔴 hay directos, refresco activo" if hay_directos and auto_refresh else ""}' f'
', unsafe_allow_html=True, ) # ── Resumen ──────────────────────────────────────────────────────────────── if not partidos: st.info("📅 Hoy no hay partidos programados para los jugadores del CSV.") st.stop() c1, c2, c3 = st.columns(3) c1.metric("🕒 Programados", len(proximos)) c2.metric("🔴 En vivo", len(en_vivo)) c3.metric("✓ Finalizados", len(finalizados)) # CSS de animación blink (compartido) st.markdown( """ """, unsafe_allow_html=True, ) # ── Render por secciones ─────────────────────────────────────────────────── if en_vivo: st.markdown("### 🔴 En vivo") for p in en_vivo: _render_partido(p) st.markdown("") if proximos: st.markdown("### 🕒 Próximos") for p in proximos: _render_partido(p) st.markdown("") if finalizados: st.markdown("### ✓ Finalizados") for p in finalizados: _render_partido(p) # ── Auto-refresh: sólo si hay directos y el toggle está activo ───────────── if hay_directos and auto_refresh: time.sleep(60) st.rerun()