Create scraping.py

scraping.py

@@ -0,0 +1,276 @@
+# ============================================================
+# 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
+
+
+# ------------------------------------------------------------
+# 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",
+]