# ============================================================ # scraping.py — Descarga de tablas web (ScraperAPI + BeautifulSoup) # y utilidades de limpieza para su procesamiento # ------------------------------------------------------------ # NOTA PARA AUDITORÍA: # - Este módulo encapsula: # (1) Limpieza básica de texto y conversión de cantidades monetarias. # (2) Descarga robusta de tablas HTML mediante ScraperAPI + BeautifulSoup. # (3) Helpers para obtener las tablas de bitcointreasuries y generar # una lista de tickers (para usar en el dropdown de la UI). # - No tiene efectos secundarios al importarlo. # - API key de ScraperAPI: # * Se lee de la variable de entorno SCRAPERAPI_KEY si no se pasa # explícitamente por parámetro. # ============================================================ from __future__ import annotations import os import re from typing import Optional, Tuple, List import numpy as np import pandas as pd import requests from bs4 import BeautifulSoup from config import ( SCRAPERAPI_KEY, CRYPTOCOMPARE_API_KEY, TIMEZONE_TARGET, GRADIO_SHARE, CRYPTOCOMPARE_BASE_URL, BITCOINTREASURIES_STOCKS_URL, BITCOINTREASURIES_ETFS_URL, YF_SEARCH_ENDPOINT, TICKERS_TO_IDS ) # ------------------------------------------------------------ # Constantes de origen (bitcointreasuries) # ------------------------------------------------------------ BITCOINTREASURIES_STOCKS_URL = "https://bitcointreasuries.net/" BITCOINTREASURIES_ETFS_URL = "https://bitcointreasuries.net/etfs-and-exchanges" # ============================================================ # Utilidades de limpieza # ============================================================ def limpiar_texto(texto: str) -> str: """ Elimina caracteres especiales y espacios extra de una cadena. Parámetros ---------- texto : str Texto de entrada. Devuelve -------- str Texto limpio (solo letras, dígitos, espacios, puntos y comas). """ # Coincide con tu lógica original: eliminar todo lo que no sea # letras/números/espacios/punto/coma con una pequeña whitelist # y luego hacer strip. texto_limpio_parcial = re.sub(r"[^a-zA-Z0-9.,\s-]", "", texto) texto_final = texto_limpio_parcial.strip() return texto_final def expand_money(val) -> float: """ Convierte representaciones monetarias abreviadas a valor numérico (float). Soporta sufijos: - 'B' (billions) → *1_000_000_000 - 'M' (millions) → *1_000_000 - 'K' (thousands) → *1_000 Parámetros ---------- val : any Cadena como '$1.2B', '500M', '40K', o valor numérico/NaN. Devuelve -------- float Valor expandido en unidades básicas (p. ej. USD). Devuelve NaN si falla. """ if pd.isna(val): return np.nan s = str(val).replace("$", "").replace(",", "").upper() try: if "B" in s: return float(s.replace("B", "")) * 1_000_000_000 if "M" in s: return float(s.replace("M", "")) * 1_000_000 if "K" in s: return float(s.replace("K", "")) * 1_000 return float(s) except Exception: return np.nan # ============================================================ # Descarga de tablas con ScraperAPI + BeautifulSoup # ============================================================ def descargar_tabla_web( url: str, indice_tabla: int = 0, api_key: Optional[str] = None, timeout: float = 30.0, ) -> Optional[pandas.core.frame.DataFrame]: """ Descarga una tabla HTML desde `url` usando ScraperAPI para evitar bloqueos. La lógica respeta tu implementación original: - Construye la URL de ScraperAPI con ?api_key=...&url=... - Parseo con BeautifulSoup - Selección de la tabla por índice (0 por defecto) - Construcción del DataFrame (con headers si están presentes) Parámetros ---------- url : str URL objetivo que contiene la/las tabla(s). indice_tabla : int Índice de la tabla a extraer en el DOM (0 = primera). api_key : str, opcional Clave de ScraperAPI. Si no se provee, se usa la variable de entorno SCRAPERAPI_KEY. Si tampoco está, la petición se realizará directamente a `url` sin pasar por ScraperAPI (puede fallar por bloqueos anti-bot). timeout : float Timeout de la petición HTTP (segundos). Devuelve -------- pandas.DataFrame | None DataFrame con los datos de la tabla, o None si no se encuentra/ocurre error. """ try: # Determinar endpoint efectivo: con o sin ScraperAPI api_key = api_key or os.getenv("SCRAPERAPI_KEY", "").strip() if api_key: api_url = f"http://api.scraperapi.com?api_key={api_key}&url={url}" target = api_url else: # Fallback directo a la URL si no hay API key target = url response = requests.get(target, timeout=timeout) response.raise_for_status() soup = BeautifulSoup(response.text, "html.parser") tables = soup.find_all("table") if len(tables) <= indice_tabla: print(f"[scraping] No se encontró la tabla con índice {indice_tabla} en {url}.") return None table = tables[indice_tabla] # Extraer cabeceras (si existen) headers = [th.text.strip() for th in table.find_all("th")] data = [] for row in table.find_all("tr"): cols = row.find_all("td") cols = [ele.text.strip() for ele in cols] if cols: data.append(cols) # Construir DataFrame con o sin headers (si la longitud coincide) if headers and data and len(headers) == len(data[0]): df = pd.DataFrame(data, columns=headers) else: df = pd.DataFrame(data) return df except requests.exceptions.RequestException as e: print(f"[scraping] Error al acceder a la URL: {e}") return None except Exception as e: print(f"[scraping] Ocurrió un error inesperado: {e}") return None # ============================================================ # Helpers específicos para bitcointreasuries # ============================================================ def get_bitcointreasuries_tables( stocks_table_index: int = 3, etfs_table_index: int = 3, api_key: Optional[str] = None, ) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]: """ Descarga las tablas de *stocks* y *ETFs* desde bitcointreasuries. Parámetros ---------- stocks_table_index : int Índice de la tabla en la página principal (por defecto: 3). etfs_table_index : int Índice de la tabla en la página de ETFs (por defecto: 3). api_key : str, opcional Clave de ScraperAPI para el scraping. Devuelve -------- (DataFrame | None, DataFrame | None) Tupla (df_stocks, df_etfs). Cada uno puede ser None si falla. """ df_stocks = descargar_tabla_web( BITCOINTREASURIES_STOCKS_URL, indice_tabla=stocks_table_index, api_key=api_key ) df_etfs = descargar_tabla_web( BITCOINTREASURIES_ETFS_URL, indice_tabla=etfs_table_index, api_key=api_key ) return df_stocks, df_etfs def generate_tickers_from_bitcointreasuries( stocks_table_index: int = 3, etfs_table_index: int = 3, api_key: Optional[str] = None, ) -> List[str]: """ Genera una lista de nombres (y típicamente tickers al final del nombre) a partir de las tablas de bitcointreasuries. NOTA: En tu `app1 6.py` original, se reutiliza la columna 'Name' de la tabla (índice 2) y luego, en otros pasos, se separa 'Name' y 'Ticker' por el último espacio. Aquí devolvemos simplemente la lista de 'Name' limpia. Parámetros ---------- stocks_table_index : int Índice de la tabla en la página principal de stocks. etfs_table_index : int Índice de la tabla en la página de ETFs. api_key : str, opcional Clave de ScraperAPI. Devuelve -------- list[str] Lista concatenada de nombres (stocks + etfs) limpios. """ tickers: List[str] = [] df_stocks, df_etfs = get_bitcointreasuries_tables( stocks_table_index=stocks_table_index, etfs_table_index=etfs_table_index, api_key=api_key, ) # Procesar stocks (si existen). En tu código original se usa la columna 2. if df_stocks is not None and not df_stocks.empty: try: df_stocks = df_stocks.copy() df_stocks["Name"] = df_stocks[2].apply(limpiar_texto) tickers.extend(df_stocks["Name"].dropna().tolist()) except Exception: # En caso de estructura inesperada de columnas pass # Procesar ETFs (si existen). En tu código original se usa la columna 2. if df_etfs is not None and not df_etfs.empty: try: df_etfs = df_etfs.copy() df_etfs["Name"] = df_etfs[2].apply(limpiar_texto) tickers.extend(df_etfs["Name"].dropna().tolist()) except Exception: pass return tickers __all__ = [ "descargar_tabla_web", "limpiar_texto", "expand_money", "get_bitcointreasuries_tables", "generate_tickers_from_bitcointreasuries", ]