Spaces:

Quant0
/

AFI

Sleeping

AFI / scraping.py

Update scraping.py

3c8b740 verified over 1 year ago

9.42 kB

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

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