Blame for scraping.py

# ============================================================
# 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",
]