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 # ----------------------------------------------- # 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}"esCount=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)