Web Scraping con Selenium de productos de Tottus.com

									import logging
									import pandas as pd
									from selenium import webdriver
									from selenium.common.exceptions import TimeoutException
									from selenium.webdriver.common.by import By
									from selenium.webdriver.edge.options import Options
									from selenium.webdriver.remote.webelement import WebElement
									from selenium.webdriver.support import expected_conditions as EC
									from selenium.webdriver.support.ui import WebDriverWait
								

									# Configurar el logging
									logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

									# Configurar el navegador
									edge_options = Options()
									edge_options.add_argument("--headless")
									edge_options.add_argument("--no-sandbox")
									edge_options.add_argument("--disable-dev-shm-usage")

									# Crear una instancia del navegador
									driver = webdriver.Edge(options=edge_options)

									# Tiempo de espera explícito (hasta 10 seg)
									wait = WebDriverWait(driver, 10)

									# Establecer el término de búsqueda, el buen precio y la URL de la página a scrapear
									termino_busqueda = "arroz"
									buen_precio = 18.0
									url_inicial = f"https://www.tottus.com.pe/tottus-pe/buscar?Ntt={termino_busqueda}"
								

									# Función para obtener el precio de un obj_producto
									def obtener_precio(elemento: WebElement, selector: str) -> float:
										elementos = elemento.find_elements(By.CSS_SELECTOR, selector)

										if elementos:
											try:
												valor = elementos[0].get_attribute(selector[3:-1])
												return float(valor.replace(",", "").strip())
											except (ValueError, AttributeError):
												return 0.0
										return 0.0
								

									resultados_totales = []
									pagina_actual = 1

									try:
										# 1. Obtener la URL inicial a escrapear
										driver.get(url_inicial)
										logging.info(f"Iniciando scraping en: {url_inicial}")

										# 2. Bucle principal
										while True:
											logging.info(f"--- Procesando página {pagina_actual} ---")
											pagina_invalida = False

											# A. Esperar que carguen los productos (buscamos el contenedor de precios como referencia)
											try:
												wait.until(EC.presence_of_all_elements_located((By.CLASS_NAME, "prices")))
											except TimeoutException:
												pagina_invalida = True

											# B. Evaluar si la página es inválida
											if pagina_invalida:
												logging.warning(f"Página {pagina_actual} marcada como inválida. Se omite extracción.")
											else:
												# C. Capturar listas de propiedades de productos
												listado_marcas = driver.find_elements(By.CLASS_NAME, "title-rebrand")
												listado_productos = driver.find_elements(By.CLASS_NAME, "subTitle-rebrand")
												listado_precios = driver.find_elements(By.CLASS_NAME, "prices")
												listado_links = driver.find_elements(By.CLASS_NAME, "pod-link")

												# D. Validar que las listas tengan el mismo tamaño
												if not (
													len(listado_marcas)
													== len(listado_productos)
													== len(listado_precios)
													== len(listado_links)
												):
													logging.error(f"Error de consistencia en página {pagina_actual}. Se omite extracción.")
												else:
													# E. Iterar sobre los productos de la página actual
													for obj_marca, obj_producto, obj_precio, obj_link in zip(
														listado_marcas,
														listado_productos,
														listado_precios,
														listado_links
													):
														try:
															marca_texto = obj_marca.text.strip()
															producto_texto = obj_producto.text.strip()
															
															precio_cmr = obtener_precio(obj_precio, "li[data-cmr-price]")
															precio_internet = obtener_precio(obj_precio, "li[data-internet-price]")
															precio_normal = obtener_precio(obj_precio, "li[data-normal-price]")

															url_producto = obj_link.get_attribute("href").strip()
															sku_producto = str(url_producto.split("/")[-1]) if url_producto else "N/A"

															# Lógica de buen precio
															precio_referencia = (
																precio_cmr
																if precio_cmr > 0
																else precio_internet
																if precio_internet > 0
																else precio_normal
															)

															es_buen_precio = "SI" if 0 < precio_referencia <= buen_precio else "NO"

															# F. Agregar los resultados a una lista
															resultados_totales.append({
																"MARCA": marca_texto,
																"PRODUCTO": producto_texto,
																"SKU": sku_producto,
																"PRECIO CMR": precio_cmr,
																"PRECIO INTERNET": precio_internet,
																"PRECIO TIENDA": precio_normal,
																"BUEN PRECIO": es_buen_precio,
																"URL": url_producto
															})

														except Exception as e:
															logging.error(f"Error al procesar producto {producto_texto}: {e}")

													logging.info(f"Cantidad de productos encontrados en la página {pagina_actual}: {len(listado_productos)}")

											# G. Lógica de paginación
											try:
												# --- Preparar para detectar cambios en la página ---

												# Guardar la URL actual
												url_anterior = driver.current_url

												elemento_viejo = None
												try:
													# Guardar el primer producto del DOM actual
													elemento_viejo = driver.find_element(By.CLASS_NAME, "prices")
												except Exception:
													logging.warning(f"No se pudo capturar el primer elemento de la página {pagina_actual}.")

												# Esperar a que el botón "Siguiente" sea cliqueable y capturarlo
												selector_boton_siguiente = "button[id='testId-pagination-bottom-arrow-right']"
												boton_siguiente = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, selector_boton_siguiente)))
												
												# --- Cambiar de página ---

												# Hacer scroll hasta que el botón esté centrado en la página para asegurar que el clic funcione
												driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", boton_siguiente)
												boton_siguiente.click()

												# --- Validar que la página ha cambiado ---

												# Esperar que la nueva URL sea diferente a la anterior
												wait.until(EC.url_changes(url_anterior))

												# Esperar que el primer elemento del DOM anterior ya no esté presente
												if elemento_viejo is not None:
													wait.until(EC.staleness_of(elemento_viejo))

												# Incrementar el contador de página
												pagina_actual += 1
												
											except TimeoutException:
												logging.info("No se detectó cambio de página o botón siguiente. Fin del scraping.")
												break
											except Exception as e:
												logging.error(f"Error al intentar cambiar de página: {e}")
												break

										# 3. Crear un DataFrame con los resultados
										df_datos = pd.DataFrame(resultados_totales)

										if not df_datos.empty:
											pd.options.display.float_format = "S/{:,.2f}".format
											df_datos.to_excel(f"productos_tottus_{termino_busqueda}.xlsx", index=False, engine="openpyxl")
											logging.info(f"Scraping finalizado. Total productos encontrados: {len(df_datos)}")
										else:
											logging.warning("No se extrajeron datos.")

									except Exception as e:
										logging.critical(f"Ocurrió un error durante la ejecución: {e}")
									finally:
										driver.quit()
										logging.info("Navegador cerrado. Ejecución finalizada.")