Create utils.py

utils.py

@@ -0,0 +1,149 @@
+import re
+import requests
+import pandas as pd
+import numpy as np
+from bs4 import BeautifulSoup
+from fuzzywuzzy import process
+from datetime import datetime, timedelta
+from config import tickers_to_ids
+from config import api_key, api_url
+
+
+#     -----------------------------------------------
+#                HERRAMIENTAS DE FROMATO 
+#     -----------------------------------------------
+
+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
+
+
+#     -----------------------------------------------
+#                 FUNCIÓN DE SCRAPER API
+#     -----------------------------------------------
+
+
+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={api_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
+
+        # El resto del código para extraer los datos de la tabla sigue igual
+        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
+
+
+#     -----------------------------------------------
+#             FUNCIÓN PARA LEER LA PLANTILLA
+#     -----------------------------------------------
+
+
+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
+
+
+#     -----------------------------------------------
+#             FUNCIÓN PARA EL CUADRO DE BÚSQUEDA
+#     -----------------------------------------------
+
+# Lista de tickers prioritarios
+priority_tickers = tickers_to_ids.copy()
+
+def find_company_online(query):
+    query_clean = query.strip()
+
+    # Petición a la API de Yahoo Finance
+    url = f"https://query1.finance.yahoo.com/v1/finance/search?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
+
+    # Lista de resultados: (nombre, ticker)
+    results = [
+        (item.get("shortname") or item.get("longname") or item.get("symbol"), item["symbol"])
+        for item in data["quotes"]
+        if "symbol" in item
+    ]
+
+    # Priorizar tickers de la lista
+    for name, ticker in results:
+        if ticker in priority_tickers:
+            return name, ticker
+
+    # Buscar coincidencia exacta/aproximada por 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
+
+    # Buscar coincidencia por nombre
+    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
+
+    # Si no hay nada claro, devolver el primer resultado
+    return results[0] if results else (None, None)
+