Create utils.py

utils.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+# Funciones auxiliares para limpieza, conversión y scraping
+
+import re
+import numpy as np
+import pandas as pd
+import requests
+from bs4 import BeautifulSoup
+from rapidfuzz import process
+from config import SCRAPERAPI_KEY, YAHOO_FINANCE_SEARCH_URL
+
+def limpiar_texto(texto):
+    """Elimina caracteres especiales y espacios extra."""
+    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):
+    """Convierte strings de dinero (ej. '$1.2B') a valores flotantes."""
+    if pd.isna(val):
+        return np.nan
+    val = str(val).replace('$', '').replace(',', '').upper()
+    try:
+        if 'B' in val:
+            return float(val.replace('B', '')) * 1_000_000_000
+        elif 'M' in val:
+            return float(val.replace('M', '')) * 1_000_000
+        elif 'K' in val:
+            return float(val.replace('K', '')) * 1_000
+        else:
+            return float(val)
+    except:
+        return np.nan
+
+def extract_file_id(url):
+    """Extrae el ID del archivo de una URL de Google Drive."""
+    match = re.search(r'/d/([a-zA-Z0-9_-]+)', url)
+    if match:
+        return match.group(1)
+    return None
+
+def descargar_tabla_web(url, indice_tabla=0):
+    """Descarga una tabla de una URL usando ScraperAPI para evitar bloqueos."""
+    api_url = f"http://api.scraperapi.com?api_key={SCRAPERAPI_KEY}&url={url}"
+    try:
+        response = requests.get(api_url)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, 'html.parser')
+        tables = soup.find_all('table')
+        if len(tables) > indice_tabla:
+            table = tables[indice_tabla]
+        else:
+            print(f"No se encontró la tabla con índice {indice_tabla} en la página: {url}.")
+            return None
+        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)
+        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"Error al acceder a la URL: {e}")
+        return None
+    except Exception as e:
+        print(f"Ocurrió un error inesperado: {e}")
+        return None
+
+def find_company_online(query, priority_tickers=None):
+    """Busca una empresa en Yahoo Finance y devuelve (nombre, ticker)."""
+    query_clean = query.strip()
+    url = f"{YAHOO_FINANCE_SEARCH_URL}?q={query_clean}&quotesCount=10&newsCount=0"
+    headers = {"User-Agent": "Mozilla/5.0"}
+    r = requests.get(url, headers=headers)
+    data = r.json()
+    if "quotes" not in data or not data["quotes"]:
+        return None, None
+    results = [
+        (item.get("shortname") or item.get("longname") or item.get("symbol"), item["symbol"])
+        for item in data["quotes"]
+        if "symbol" in item
+    ]
+    if priority_tickers:
+        for name, ticker in results:
+            if ticker in priority_tickers:
+                return name, ticker
+    tickers = [sym for _, sym in results]
+    best_ticker_match = process.extractOne(query_clean.upper(), tickers, score_cutoff=80)
+    if best_ticker_match:
+        ticker = best_ticker_match[0]
+        name = next(name for name, sym in results if sym == ticker)
+        return name, ticker
+    names = [name for name, _ in results]
+    best_name_match = process.extractOne(query_clean, names, score_cutoff=60)
+    if best_name_match:
+        name = best_name_match[0]
+        ticker = next(sym for nm, sym in results if nm == name)
+        return name, ticker
+    return results[0] if results else (None, None)