"""Proveedor de datos REAL vía Scweet (https://github.com/Altimis/Scweet). Usa la GraphQL interna de X con la cookie `auth_token` de una cuenta SECUNDARIA (nunca tu cuenta personal). Scweet gestiona internamente el token anti-bot 'x-client-transaction-id', por eso funciona donde twikit fallaba, y trae datos FRESCOS de cualquier cuenta (resuelve el caché viejo de syndication). La cookie se lee, por orden: 1) variable de entorno SCWEET_AUTH_TOKEN 2) archivo de secretos local (cfg['scweet_secrets_file'], por defecto secrets.local.json) 3) cfg['scweet_auth_token'] (desaconsejado: no lo pongas en un config.yaml versionado) Guía para obtener la cookie: ver SETUP_SCWEET.md. """ import json import os import time from datetime import datetime, timezone from typing import List, Optional from agent.models import Tweet from agent.providers.base import TweetProvider class ScweetError(Exception): pass def _to_int(value) -> int: """Convierte 1234, '1,234', '1.2K', '3.4M' -> int.""" if value is None: return 0 if isinstance(value, (int, float)): return int(value) s = str(value).strip().upper().replace(",", "") if not s: return 0 mult = 1 if s.endswith("K"): mult, s = 1_000, s[:-1] elif s.endswith("M"): mult, s = 1_000_000, s[:-1] elif s.endswith("B"): mult, s = 1_000_000_000, s[:-1] try: return int(float(s) * mult) except ValueError: return 0 def _parse_ts(value) -> Optional[datetime]: if value is None: return None if isinstance(value, datetime): return value if value.tzinfo else value.replace(tzinfo=timezone.utc) if isinstance(value, (int, float)): v = float(value) if v > 1e12: # milisegundos v /= 1000.0 return datetime.fromtimestamp(v, tz=timezone.utc) s = str(value).strip() try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except ValueError: pass for fmt in ("%a %b %d %H:%M:%S %z %Y", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): try: dt = datetime.strptime(s, fmt) return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) except ValueError: continue return None def _first(row: dict, *keys): for k in keys: if row.get(k) not in (None, ""): return row.get(k) return None def _author_of(row: dict, fallback: str) -> str: u = row.get("user") if isinstance(u, dict): return u.get("screen_name") or u.get("name") or fallback if isinstance(u, str) and u: return u.lstrip("@") return fallback def row_to_tweet(row: dict, handle: str) -> Optional[Tweet]: """Mapea un dict de Scweet (TweetRecord) a nuestro modelo Tweet. Tolerante a variantes de claves.""" created = _parse_ts(_first(row, "timestamp", "date", "created_at")) if created is None: return None author = _author_of(row, handle.lstrip("@")) raw = row.get("raw") or {} tid = str(_first(row, "tweet_id", "id", "id_str") or raw.get("id_str") or "") url = _first(row, "tweet_url", "url") or (f"https://x.com/{author}/status/{tid}" if tid else "") return Tweet( id=tid, author=author, text=_first(row, "text", "embedded_text", "full_text") or "", created_at=created, likes=_to_int(_first(row, "likes", "favorite_count", "like_count")), retweets=_to_int(_first(row, "retweets", "retweet_count")), replies=_to_int(_first(row, "comments", "reply_count", "replies")), quotes=_to_int(raw.get("quote_count")), views=_to_int(_first(row, "views", "view_count") or raw.get("view_count")), url=url, ) class ScweetProvider(TweetProvider): def __init__(self, cfg: dict = None): cfg = cfg or {} self.cfg = cfg self.limit = int(cfg.get("scweet_limit", 50)) self.db_path = cfg.get("scweet_db_path", "data/scweet_state.db") self.secrets_file = cfg.get("scweet_secrets_file", "secrets.local.json") self.manifest_scrape = bool(cfg.get("scweet_manifest_scrape_on_init", False)) self.delay = float(cfg.get("scweet_delay_seconds", 2)) self._client = None self._last_request = 0.0 self.last_meta = {} def _auth_token(self) -> str: tok = os.environ.get("SCWEET_AUTH_TOKEN") if tok: return tok.strip() path = self.secrets_file if not os.path.isabs(path): path = os.path.join(os.getcwd(), path) if os.path.exists(path): try: with open(path, encoding="utf-8") as f: data = json.load(f) if data.get("scweet_auth_token"): return str(data["scweet_auth_token"]).strip() except Exception: pass if self.cfg.get("scweet_auth_token"): return str(self.cfg["scweet_auth_token"]).strip() raise ScweetError( "Falta la cookie auth_token. Define SCWEET_AUTH_TOKEN o crea " f"'{self.secrets_file}' con {{\"scweet_auth_token\": \"...\"}}. Ver SETUP_SCWEET.md." ) def _get_client(self): if self._client is None: try: from Scweet.client import Scweet as ScweetClient except Exception as e: raise ScweetError(f"No se pudo importar Scweet: {e}") token = self._auth_token() db = self.db_path if not os.path.isabs(db): db = os.path.join(os.getcwd(), db) os.makedirs(os.path.dirname(db) or ".", exist_ok=True) self._client = ScweetClient( auth_token=token, db_path=db, manifest_scrape_on_init=self.manifest_scrape, ) return self._client def _throttle(self) -> None: wait = self.delay - (time.monotonic() - self._last_request) if wait > 0: time.sleep(wait) self._last_request = time.monotonic() def fetch_user_tweets(self, handle: str, since: datetime, limit: int = 50) -> List[Tweet]: client = self._get_client() self._throttle() try: rows = client.get_profile_tweets([handle.lstrip("@")], limit=limit or self.limit) except ScweetError: raise except Exception as e: raise ScweetError(f"Scweet falló en @{handle}: {e}") tweets = [t for t in (row_to_tweet(r, handle) for r in (rows or [])) if t] newest = max((t.created_at for t in tweets), default=None) self.last_meta[handle.lstrip("@")] = {"newest": newest, "stale": False, "n_raw": len(tweets)} tweets = [t for t in tweets if t.created_at >= since] tweets.sort(key=lambda t: t.created_at, reverse=True) return tweets[:limit]