"""Dashboard del agente de Twitter/X (Streamlit). Ejecutar con: streamlit run app.py """ import os from datetime import datetime, timedelta, timezone import streamlit as st from agent.config import load_config from agent.models import Account from agent.ranking import rank_tweets from agent.registry import get_provider, get_summarizer from agent.store import load_accounts, save_accounts HERE = os.path.dirname(os.path.abspath(__file__)) st.set_page_config(page_title="Twitter Agent", page_icon="🐦", layout="wide") cfg = load_config(os.path.join(HERE, "config.yaml")) if not os.path.isabs(cfg["accounts_path"]): cfg["accounts_path"] = os.path.join(HERE, cfg["accounts_path"]) for _k in ("scweet_db_path", "scweet_secrets_file", "cache_dir"): if cfg.get(_k) and not os.path.isabs(cfg[_k]): cfg[_k] = os.path.join(HERE, cfg[_k]) # ---------- helpers ---------- def humanize_age(dt: datetime) -> str: secs = (datetime.now(timezone.utc) - dt).total_seconds() if secs < 3600: return f"hace {int(secs // 60)} min" if secs < 86400: return f"hace {int(secs // 3600)} h" return f"hace {int(secs // 86400)} d" @st.cache_data(show_spinner="Capturando tweets… (con datos reales puede tardar unos segundos por cuenta)") def fetch_all(handles, provider_name, window_hours): """Devuelve (tweets, status). status[handle] = {'count': n, 'error': str|None}.""" provider = get_provider(provider_name, cfg) if cfg.get("cache_enabled", True) and provider_name != "mock": from agent.providers.cached import CachedProvider provider = CachedProvider(provider, cfg.get("cache_dir", "data/cache"), cfg.get("cache_ttl_minutes", 60)) since = datetime.now(timezone.utc) - timedelta(hours=window_hours) tweets, status = [], {} for h in handles: try: tw = provider.fetch_user_tweets(h, since=since, limit=50) tweets.extend(tw) meta = getattr(provider, "last_meta", {}).get(h.lstrip("@"), {}) status[h] = { "count": len(tw), "error": None, "stale": bool(meta.get("stale", False)), "newest": meta.get("newest"), "n_raw": int(meta.get("n_raw", len(tw))), } except Exception as e: # una cuenta concreta puede fallar (rate-limit, etc.) status[h] = {"count": 0, "error": str(e), "stale": False, "newest": None, "n_raw": 0} return tweets, status # ---------- estado ---------- if "accounts" not in st.session_state: st.session_state.accounts = load_accounts(cfg["accounts_path"]) # ---------- barra lateral ---------- with st.sidebar: st.title("🐦 Twitter Agent") st.caption("Captura · Ranking · Resumen de tus cuentas") provider_name = st.selectbox( "Fuente de datos", ["scweet", "syndication"], help=( "scweet = tweets REALES y frescos de CUALQUIER cuenta (requiere cookie de una " "cuenta secundaria; ver SETUP_SCWEET.md). " "syndication = tweets REALES sin login (gratis; alguna cuenta puede venir 'vieja')." ), ) st.divider() st.subheader("⚙️ Ajustes del TOP") window_hours = st.slider("Ventana (horas)", 6, 168, int(cfg["window_hours"]), step=6) top_n = st.slider("Cuántos TOP", 5, 30, int(cfg["top_n"])) halflife = st.slider( "Peso de la frescura (vida media, h)", 6, 72, int(cfg["recency_halflife_hours"]), help="Más bajo = prioriza lo más reciente. Más alto = prioriza el engagement.", ) topics_str = st.text_input("Temas a potenciar (separa por comas)", ", ".join(cfg["topics"])) topics = [t.strip() for t in topics_str.split(",") if t.strip()] if st.button("🔄 Actualizar datos", use_container_width=True): st.cache_data.clear() _cdir = cfg.get("cache_dir") if _cdir and os.path.isdir(_cdir): for _fn in os.listdir(_cdir): if _fn.endswith(".json"): try: os.remove(os.path.join(_cdir, _fn)) except OSError: pass st.rerun() enabled = [a for a in st.session_state.accounts if a.enabled] handles = [a.handle for a in enabled] # Captura + ranking una sola vez (cacheado); lo usan las pestañas TOP y Resumen. tweets, status, ranked = [], {}, [] if handles: tweets, status = fetch_all(handles, provider_name, window_hours) ranked = rank_tweets( tweets, top_n=top_n, recency_halflife_hours=halflife, keywords=topics, weights=cfg["weights"], ) tab_top, tab_summary, tab_accounts = st.tabs(["🔝 TOP", "📝 Resumen", "📋 Cuentas"]) # ---------- TOP ---------- with tab_top: st.header("🔝 Tweets TOP") if not handles: st.info("No hay cuentas activas. Añade o activa alguna en la pestaña **Cuentas**.") else: errs = [h for h, s in status.items() if s.get("error")] stale = [h for h, s in status.items() if not s.get("error") and s.get("stale") and s.get("count", 0) == 0] ratelimited = [h for h, s in status.items() if not s.get("error") and not s.get("stale") and s.get("count", 0) == 0 and s.get("n_raw", 1) == 0] quiet = [h for h, s in status.items() if not s.get("error") and not s.get("stale") and s.get("count", 0) == 0 and s.get("n_raw", 1) > 0] if errs: st.warning( "⚠️ No respondieron: " + ", ".join("@" + h for h in errs) + ". Suele ser rate-limit — reintenta en unos minutos." ) if ratelimited: st.warning( "🚦 Sin datos (probable rate-limit de la cuenta de Scweet): " + ", ".join("@" + h for h in ratelimited) + ". Deja descansar la cuenta unos minutos, o añade una 2ª cuenta secundaria." ) if stale: st.info( "🟠 X sirve caché antiguo (proveedor 'syndication') para: " + ", ".join("@" + h for h in stale) ) if quiet: st.caption( "Sin novedades en la ventana (prueba a subir las horas): " + ", ".join("@" + h for h in quiet) ) with_data = len([h for h, s in status.items() if s.get("count", 0) > 0]) st.caption( f"{len(tweets)} tweets · {with_data}/{len(handles)} cuentas con datos · " f"últimas {window_hours} h · TOP {len(ranked)}" ) for i, (score, t) in enumerate(ranked, 1): with st.container(border=True): c1, c2 = st.columns([0.06, 0.94]) c1.markdown(f"### {i}") with c2: st.markdown(f"**@{t.author}** · {humanize_age(t.created_at)} · `score {score}`") st.write(t.text) st.markdown( f"❤️ {t.likes:,} · 🔁 {t.retweets:,} · 💬 {t.replies:,} · " f"👁️ {t.views:,} · [Abrir en X →]({t.url})" ) # ---------- Resumen ---------- with tab_summary: st.header("📝 Resumen del TOP") if not handles: st.info("Añade cuentas para generar el resumen.") else: summarizer = get_summarizer(cfg["summarizer"], cfg) st.markdown(summarizer.summarize(ranked, topics=topics)) # ---------- Cuentas ---------- with tab_accounts: st.header("📋 Cuentas vigiladas") st.caption("Activa/desactiva las que quieras incluir en el TOP, o añade nuevas.") for a in st.session_state.accounts: c1, c2, c3 = st.columns([0.34, 0.51, 0.15]) c1.markdown(f"**@{a.handle}**") c2.caption(f"{a.display_name} — {a.note}" if a.note else a.display_name) a.enabled = c3.toggle("Activa", value=a.enabled, key=f"en_{a.handle}", label_visibility="collapsed") st.divider() col_add, col_btn = st.columns([0.7, 0.3]) new_handle = col_add.text_input("Añadir cuenta (handle sin @)", key="new_handle", label_visibility="collapsed", placeholder="ej. sama") if col_btn.button("➕ Añadir", use_container_width=True): h = new_handle.lstrip("@").strip() existing = {x.handle.lower() for x in st.session_state.accounts} if h and h.lower() not in existing: st.session_state.accounts.append(Account(handle=h, display_name=h)) save_accounts(cfg["accounts_path"], st.session_state.accounts) st.cache_data.clear() st.rerun() elif h: st.warning("Esa cuenta ya está en la lista.") if st.button("💾 Guardar cambios"): save_accounts(cfg["accounts_path"], st.session_state.accounts) st.cache_data.clear() st.success("Lista de cuentas guardada.")