⚡ Web Scraping en C# con Kafka – Procesando datos en tiempo real

¿Quieres aprender a crear un scraper eficiente y escalable en C#? En este tutorial, te guiaré paso a paso para construir desde cero un sistema de Web Scraping utilizando Selenium junto a Apache Kafka para la gestión de eventos.

He diseñado este template con un enfoque completamente didáctico, pensado para que comprendas cada parte del proceso: desde cómo enviar URLs a una cola de eventos con Kafka hasta cómo consumirlas y procesarlas de forma asíncrona.

🎁 Y lo mejor de todo… ¡es totalmente gratis!

Aquí aprenderás a escalar aplicaciones de scraping usando colas de eventos y a procesar datos de forma distribuida, todo con ejemplos claros y un enfoque práctico.

---

🚀 ¿Qué es Apache Kafka y por qué usarlo aquí?

Apache Kafka es una plataforma distribuida de mensajería orientada a flujos de datos. Funciona como un sistema de colas altamente escalable y es ideal para procesar grandes volúmenes de eventos en tiempo real.

🏆 Ventajas de usar Kafka en un proyecto de Web Scraping:

• Escalabilidad: Puedes agregar más consumidores o productores fácilmente.
• Tolerancia a fallos: Kafka almacena los mensajes de forma segura hasta que sean procesados.
• Desacoplamiento: El scraper (productor) y el consumidor están completamente desacoplados.
• Procesamiento en tiempo real: Permite manejar eventos a medida que ocurren.

---

📁 Estructura del Proyecto

He estructurado el proyecto para centrarme en los aspectos más importantes y didácticos, simplificando el código y eliminando cualquier complejidad innecesaria. El objetivo es que puedas entender fácilmente cómo funciona cada parte del sistema y cómo se integran entre sí.

InmoCopilotoScrapper/
├── Application/
│   └── Events/
│       └── ScrapperConsumerService.cs  # Servicio que consume los eventos desde Kafka
├── Domain/
│   ├── Controllers/
│   │   └── ScrappingController.cs      # Controlador que envía eventos a Kafka
│   ├── Entities/
│   │   └── WebPageContent.cs           # Entidad que representa el contenido extraído
│   └── Interfaces/
│       └── IEventRepository.cs         # Interfaz para manejar eventos
├── Infrastructure/
│   └── EventRepository.cs              # Implementación de la interfaz para Kafka
├── Services/
│   └── WebScrapingService.cs           # Servicio que realiza el scraping
├── appsettings.json                    # Configuración general
├── appsettings.Development.json        # Configuración para entorno de desarrollo
└── Program.cs                          # Entrada principal del programa

Este enfoque modular te permitirá entender claramente cada parte del flujo y cómo Kafka se integra en el proceso.

---

🧩 Flujo de Trabajo – ¿Cómo funciona?

1. El usuario hace una solicitud al ScrappingController.
2. El controlador envía un evento a Kafka, indicando que hay una URL que necesita ser scrapeada.
3. ScrapperConsumerService actúa como consumidor, escuchando eventos desde el topic de Kafka.
4. Cuando llega un evento, el consumidor ejecuta el WebScrapingService para realizar el scraping de la URL proporcionada.
5. El resultado se procesa y almacena, o se envía a otro microservicio.

---

🔗 Conectando C# con Kafka

1️⃣ Instalando dependencias

Primero, asegúrate de tener las dependencias necesarias en tu proyecto. Para conectarte a Kafka desde C#, puedes usar la librería Confluent.Kafka.

dotnet add package Confluent.Kafka

---

2️⃣ Productor Kafka – ScrappingController.cs

El controlador será el encargado de recibir peticiones HTTP y enviar los datos a Kafka.

using InmoCopilotoScrapper.Domain.Interfaces;
using Microsoft.AspNetCore.Mvc;

namespace InmoCopilotoScrapper.Controllers;
[ApiController]
[Route("[controller]")]
public class ScrapingController : ControllerBase
{
    private readonly IEventRepository _eventRepository;

    public ScrapingController(IEventRepository eventRepository)
    {
        _eventRepository = eventRepository;
    }

    [HttpGet("scrape")]
    public async Task<IActionResult> ScrapeWebsite([FromQuery] string url)
    {
        if (string.IsNullOrEmpty(url))
        {
            return BadRequest("URL is required.");
        }

        await _eventRepository.CreateScrappingEventAsync(url);

        return Ok();
    }
}

✅ ¿Qué hace este código?

• Recibe una URL desde el cuerpo de la petición.
• Envía la URL al topic de Kafka llamado scraping-topic.

---

3️⃣ Consumidor Kafka – ScrapperConsumerService.cs

El consumidor escuchará el topic y procesará las URLs.

using Confluent.Kafka;
using InmoCopilotoScrapper.Services;

namespace InmoCopilotoScrapper.Application.Events;

public class ScrapperConsumerService : BackgroundService
{
    private readonly IConsumer<Ignore, string> _consumer;
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<ScrapperConsumerService> _logger;
    private readonly IConfiguration _configuration;

    public ScrapperConsumerService(IConfiguration configuration, ILogger<ScrapperConsumerService> logger, IServiceProvider serviceProvider)
    {
        _logger = logger;

        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = configuration["Kafka:BootstrapServers"],
            GroupId = "csharp-group-1",
            AutoOffsetReset = AutoOffsetReset.Latest,
            SecurityProtocol = SecurityProtocol.SaslSsl,
            SaslMechanism = SaslMechanism.Plain,
            SaslUsername = configuration["Kafka:SaslUsername"],
            SaslPassword = configuration["Kafka:SaslPassword"],
            SessionTimeoutMs = 3600000,
            EnableAutoCommit = false, // Deshabilitar commit automático
            StatisticsIntervalMs = 60000,
            EnablePartitionEof = true,
            MaxPollIntervalMs = 3600000,
            SocketKeepaliveEnable = true,
            MetadataMaxAgeMs = 300000,
            ReconnectBackoffMs = 1000,
            ReconnectBackoffMaxMs = 10000,
        };

        _consumer = new ConsumerBuilder<Ignore, string>(consumerConfig)
          .SetLogHandler((_, logMessage) => _logger.LogDebug(logMessage.Message))
          .SetErrorHandler((_, e) => _logger.LogError($"Error: {e.Reason}"))
          .Build();

        _serviceProvider = serviceProvider;
        _configuration = configuration;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _consumer.Subscribe(_configuration["KAFKA_TOPIC_GENERATE_SCRAPPING"]);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ProcessKafkaMessage(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError($"Error in ExecuteAsync: {ex.Message}");
            }

            await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken); // Asegúrate de no hacer delays largos aquí.
        }

        _consumer.Close();
    }

    public async Task ProcessKafkaMessage(CancellationToken stoppingToken)
    {
        string website = string.Empty;
        try
        {
            var consumeResult = _consumer.Consume(stoppingToken);

            if (consumeResult != null && consumeResult.Message != null)
            {
                website = consumeResult.Message.Value;

                using (var scope = _serviceProvider.CreateScope())
                {
                    var scrappingService = scope.ServiceProvider.GetRequiredService<WebScrapingService>();

                    var result = await scrappingService.ScrapePage(website);

                    //TODO: Haz aquí lo que necesites con el resultado del scrapping

                    _logger.LogInformation("Scrapping has been processed: {website}", website);

                    try
                    {
                        _consumer.Commit(consumeResult);
                        _logger.LogInformation("Committed offset for processed scrapping: {website}", website);
                    }
                    catch (KafkaException ex)
                    {
                        _logger.LogError($"Error committing offset: {ex.Message}");
                    }
                }
            }
            else
            {
                _logger.LogInformation("No message received in the last 10 seconds");
            }
        }
        catch (ConsumeException ex)
        {
            _logger.LogError($"Consume error: {ex.Error.Reason}");
            // Opción de manejar errores de consumo específicos aquí
        }
        catch (KafkaException ex)
        {
            _logger.LogError($"Kafka error: {ex.Error.Reason}");
            // Opción de manejar errores de Kafka específicos aquí
        }
        catch (Exception ex)
        {
            _logger.LogError($"Error processing Kafka message: {website} {ex.Message}");
            // Asegurarse de que se manejen todos los errores y se liberen recursos si es necesario
        }
    }
}

✅ ¿Qué hace este código?

• Se conecta al broker de Kafka y escucha el topic.
• Por cada URL recibida, ejecuta el scraper.
• Muestra los resultados por consola (o los guarda en una base de datos).

---

4️⃣ Servicio de Scraping – WebScrapingService.cs

Este servicio se mantiene como antes, usando Selenium para realizar el scraping. Aquí solo ejecutará el proceso para cada URL que llega desde Kafka.

using OpenQA.Selenium;
using OpenQA.Selenium.Chrome;
using System.Net;
using System.Text.RegularExpressions;
using System.Xml;

namespace InmoCopilotoScrapper.Services
{
    public class WebScrapingService
    {
        private readonly ILogger<WebScrapingService> _logger;
        private readonly ChromeDriver _driver;

        public WebScrapingService(ILogger<WebScrapingService> logger)
        {
            _logger = logger;
            var options = new ChromeOptions();

            options.AddArgument("--headless"); // Puedes desactivar temporalmente para depuración
            options.AddArgument("--disable-gpu");
            options.AddArgument("--no-sandbox");
            options.AddArgument("--disable-dev-shm-usage");

            // Establecer el idioma preferido del navegador
            options.AddArgument("--lang=es-ES");
            options.AddArgument("–-lang=es-ES");
            options.AddArgument("--accept-lang=es-ES");
            options.AddUserProfilePreference("intl.accept_languages", "es-ES,es");

            options.AddArgument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.121 Safari/537.36");

            // Emular un dispositivo móvil específico
            //options.EnableMobileEmulation("iPhone X");

            // Configuración de geolocalización (opcional)
            var locationContext = new Dictionary<string, object>
            {
                { "latitude", 40.416775 },  // Coordenadas de Madrid, España
                { "longitude", -3.703790 },
                { "accuracy", 1 }
            };
            _driver = new ChromeDriver(options);
            _driver.Manage().Cookies.DeleteAllCookies();
            _driver.ExecuteCdpCommand("Emulation.setGeolocationOverride", locationContext);
        }

        public async Task<List<PropertyData>> ScrapePage(string url)
        {
            var allUrls = new HashSet<string>();
            var propertyDataList = new List<PropertyData>();

            await Task.Run(() => GetAllLinks(url, allUrls));
            foreach (var link in allUrls)
            {
                _logger.LogInformation($"Processing link: {link}");
                var propertyData = await Task.Run(() => GetPropertyDataFromPage(link));
                if (propertyData != null)
                {
                    propertyDataList.Add(propertyData);
                }
            }

            //SAVE FIREBASE
            return propertyDataList;
        }

        private void GetAllLinks(string url, HashSet<string> allUrls)
        {
            _driver.Navigate().GoToUrl(url);

            // Extraer enlaces de la página
            var links = _driver.FindElements(By.CssSelector("a[href]"))
                .Select(element => element.GetAttribute("href"))
                .Where(href => !string.IsNullOrEmpty(href) && Uri.IsWellFormedUriString(href, UriKind.RelativeOrAbsolute))
                .Select(href => new Uri(new Uri(url), href).ToString())
                .Distinct()
                .ToList();

            _logger.LogInformation($"Found {links.Count} links on {url}");

            // Agregar links encontrados a la lista final
            foreach (var link in links)
            {
                _logger.LogInformation($"Adding link: {link}");
                allUrls.Add(link);
            }

            // Buscar el archivo sitemap.xml y extraer URLs adicionales
            var sitemapUrl = new Uri(new Uri(url), "/sitemap.xml").ToString();
            var newUrlsFromSitemap = GetUrlsFromSitemap(sitemapUrl);

            foreach (var sitemapLink in newUrlsFromSitemap)
            {
                if (allUrls.Add(sitemapLink))
                {
                    _logger.LogInformation($"Adding new link from sitemap: {sitemapLink}");
                }
            }
        }

        private List<string> GetUrlsFromSitemap(string sitemapUrl)
        {
            var newUrls = new List<string>();

            try
            {
                using (var webClient = new WebClient())
                {
                    var sitemapContent = webClient.DownloadString(sitemapUrl);
                    var xmlDoc = new XmlDocument();
                    xmlDoc.LoadXml(sitemapContent);

                    XmlNamespaceManager nsmgr = new XmlNamespaceManager(xmlDoc.NameTable);
                    nsmgr.AddNamespace("sm", "http://www.sitemaps.org/schemas/sitemap/0.9");

                    var locNodes = xmlDoc.SelectNodes("//sm:url/sm:loc", nsmgr);

                    if (locNodes != null)
                    {
                        foreach (XmlNode locNode in locNodes)
                        {
                            if (Uri.IsWellFormedUriString(locNode.InnerText, UriKind.Absolute))
                            {
                                newUrls.Add(locNode.InnerText.Trim());
                            }
                        }
                    }

                    _logger.LogInformation($"Found {newUrls.Count} links in sitemap");
                }
            }
            catch (WebException ex)
            {
                _logger.LogError($"Error accessing sitemap at {sitemapUrl}: {ex.Message}");
            }
            catch (XmlException ex)
            {
                _logger.LogError($"Error parsing sitemap XML from {sitemapUrl}: {ex.Message}");
            }

            return newUrls;
        }

        private string GetPrice()
        {
            try
            {
                // Intenta encontrar el precio utilizando XPath para clases o IDs que contienen 'precio' o 'price'
                IWebElement priceElement = _driver.FindElement(By.XPath("//*[contains(@class, 'precio') or contains(@id, 'precio') or contains(@class, 'price') or contains(@id, 'price')]"));

                // Obtén el texto del elemento
                string price = priceElement.Text?.Trim();

                // Verifica si se ha encontrado un precio visible
                if (!string.IsNullOrEmpty(price))
                {
                    return CleanPrice(price);
                }
            }
            catch (NoSuchElementException)
            {
                // No se encontró el elemento con FindElement
            }

            // Intenta buscar el precio utilizando un selector CSS como fallback
            var priceElements = _driver.FindElements(By.CssSelector("*[id*='price'], *[id*='precio'], *[class*='price'], *[class*='precio']"));

            foreach (var element in priceElements)
            {
                string? price = element.Text?.Trim();

                // Verifica si el texto del elemento no está vacío
                if (!string.IsNullOrEmpty(price))
                {
                    return CleanPrice(price);
                }

                // Intenta obtener el precio desde el atributo "content" si existe
                price = element.GetAttribute("content")?.Trim();
                if (!string.IsNullOrEmpty(price))
                {
                    return CleanPrice(price);
                }
            }

            // Si no se encontró ningún precio, devolver un mensaje indicando que no se encontró
            return "Precio no encontrado";
        }

        private string CleanPrice(string rawPrice)
        {
            // Expresión regular para encontrar números y símbolos de monedas comunes (€, $, etc.)
            var regex = new Regex(@"[\d.,]+(?:\s*[€$])?");
            var match = regex.Match(rawPrice);

            if (match.Success)
            {
                return match.Value.Trim();
            }

            return "Precio no encontrado";
        }

        private PropertyData GetPropertyDataFromPage(string url)
        {
            try
            {
                _driver.Navigate().GoToUrl(url);

                // Capturar la URL de la imagen desde la metaetiqueta og:image
                var imageUrl = _driver.FindElement(By.CssSelector("meta[property='og:image']"))?.GetAttribute("content")?.Trim();

                // Buscar la descripción utilizando varias estrategias
                var shortDescription = GetShortDescription();
                var longDescription = GetLongDescription();

                var price = GetPrice();

                var characteristics = _driver.FindElements(By.CssSelector(".caracteristicas, .features, .specs"))
                                              .Select(element => element.Text.Trim())
                                              .ToList();

                return new PropertyData
                {
                    Url = url,
                    ImageUrl = imageUrl,
                    ShortDescription = shortDescription,
                    LongDescription = longDescription,
                    Price = price,
                    Characteristics = characteristics
                };
            }
            catch (NoSuchElementException ex)
            {
                _logger.LogError($"Element not found on URL {url}: {ex.Message}");
            }
            catch (Exception ex)
            {
                _logger.LogError($"Error processing URL {url}: {ex.Message}");
            }
            return null;
        }

        private string GetShortDescription()
        {
            IWebElement metaDescription = _driver.FindElement(By.XPath("//meta[@name='description']"));
            string descriptionContent = metaDescription.GetAttribute("content");
            // Intentar encontrar la descripción usando diferentes métodos
            var description = _driver.FindElements(By.CssSelector("meta[name='description'], meta[property='og:description'], meta[name='twitter:description']"))
                .Select(meta => meta.GetAttribute("content"))
                .FirstOrDefault(content => !string.IsNullOrEmpty(content))?.Trim();

            return description ?? "Descripción no encontrada";
        }

        private string GetLongDescription()
        {
            // Intentar encontrar la descripción usando diferentes métodos
            var description = string.Empty;

            description = _driver.FindElements(By.CssSelector("section[id*='descripcion'], section[id*='description'], section[id*='desc']"))
                   .Select(section => section.Text)
                   .FirstOrDefault(text => !string.IsNullOrEmpty(text))?.Trim();

            if (string.IsNullOrEmpty(description))
            {
                // Buscar en secciones con clases relacionadas con descripciones
                description = _driver.FindElements(By.CssSelector("*[class*='descripcion'], *[class*='description'], *[class*='desc']"))
                    .Select(div => div.Text)
                    .FirstOrDefault(text => !string.IsNullOrEmpty(text))?.Trim();
            }

            return description ?? "Descripción no encontrada";
        }

        ~WebScrapingService()
        {
            _driver.Quit();
        }
    }

    public class PropertyData
    {
        public string Url { get; set; }
        public string ImageUrl { get; set; } // Nueva propiedad para almacenar la URL de la imagen
        public string ShortDescription { get; set; }
        public string LongDescription { get; set; }
        public string Price { get; set; }
        public List<string> Characteristics { get; set; }
    }
}

---

⚙️ Configuración – appsettings.json

Aquí defines los parámetros para conectar con Kafka.

{
  "Kafka": {
    "BootstrapServers": "localhost:9092",
    "ScrapingTopic": "scraping-topic"
  }
}

---

🧪 Probando el sistema

1️⃣ Levanta tu entorno de Kafka (usando Docker):

docker-compose up -d

2️⃣ Lanza el consumidor en tu aplicación:

dotnet run

3️⃣ Envía una URL al endpoint:

curl -X POST http://localhost:5000/api/scrapping/scrape -H "Content-Type: application/json" -d "\"https://www.ejemplo.com\""

Verás en la consola del consumidor algo como:

[Kafka] URL recibida: https://www.ejemplo.com
Iniciando scraping para: https://www.ejemplo.com
Scraping completado para https://www.ejemplo.com con 1 elementos encontrados

Puedes añadir el siguiente párrafo al README.md o al post:

---

📂 El proyecto completo está disponible en GitHub

Todo el código fuente de InmoCopilotoScrapper está disponible en mi repositorio de GitHub. Puedes clonarlo, revisarlo y modificarlo según tus necesidades. ¡No olvides dejar una estrella ⭐ si te resulta útil!

👉 Accede al repositorio aquí

---

¿Te gustaría que añadiera alguna instrucción específica sobre cómo clonar el repositorio o contribuir al proyecto? 🚀

---

📊 ¿Qué se puede hacer con esto?

• 🏘️ Scraping inmobiliario masivo: Extrae datos de múltiples sitios a la vez usando varios consumidores.
• 🛒 Comparadores de precios: Procesa productos en tiempo real desde distintas tiendas.
• 🔍 Análisis de tendencias: Monitorea cambios de precios o características en tiempo real.
• 📰 Feed de noticias personalizado: Extrae titulares desde múltiples fuentes.

---

🚀 Próximos pasos

• ✅ Implementar almacenamiento en una base de datos.
• ✅ Añadir lógica de reintentos si el scraping falla.
• ✅ Escalar usando múltiples consumidores en paralelo.
• ✅ Añadir autenticación y control de acceso al API.

---

📢 ¿Te gustaría aprender más?

¡Déjame un comentario! ¿Quieres que haga una guía paso a paso para desplegar esto en Kubernetes? ¿O prefieres un tutorial para implementar almacenamiento en tiempo real en Firebase?

---

👉 Sígueme para más contenido técnico:

• 🌐 Blog
• 🐦 Instagram
• 💼 LinkedIn

---

¡Nos vemos en el próximo post! 🚀🚀

---

¿Te gustaría que agregue algún fragmento adicional o amplíe algún punto?