Rastreador multiproceso en Python

En este artículo, describiremos cómo es posible construir un rastreador simple basado en subprocesos múltiples utilizando Python.

Módulos necesarios

bs4 : Beautiful Soup (bs4) es una biblioteca de Python para extraer datos de archivos HTML y XML. Para instalar esta biblioteca, escriba el siguiente comando en IDE/terminal.

pip install bs4

requests: esta biblioteca le permite enviar requests HTTP/1.1 muy fácilmente. Para instalar esta biblioteca, escriba el siguiente comando en IDE/terminal.

pip install requests

Implementación paso a paso

Paso 1: primero importaremos todas las bibliotecas que necesitamos rastrear. Si usa Python3, ya debería tener todas las bibliotecas excepto BeautifulSoup, Requests. Entonces, si aún no ha instalado estas dos bibliotecas, deberá instalarlas usando los comandos especificados anteriormente.

import multiprocessing
from bs4 import BeautifulSoup
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urljoin, urlparse
import requests

Paso 2: Cree un programa principal y luego cree un objeto de clase MultiThreadedCrawler y pase la URL inicial a su constructor parametrizado y llame al método run_web_scrawler().

if __name__ == '__main__':
    cc = MultiThreadedCrawler("https://www.geeksforgeeks.org/")
    cc.run_web_crawler()
    cc.info()

Paso 3: Cree una clase llamada MultiThreadedCrawler. E inicialice todas las variables en el constructor, asigne la URL base a la variable de instancia denominada seed_url. Y luego formatee la URL base en URL absoluta, utilizando esquemas como HTTPS y ubicación de red.

Para ejecutar la tarea de la frontera de rastreo al mismo tiempo, use subprocesos múltiples en python. Cree un objeto de la clase ThreadPoolExecutor y configure el máximo de trabajadores como 5, es decir, para ejecutar 5 subprocesos a la vez. Y para evitar visitas duplicadas a páginas web, para mantener el historial cree una estructura de datos establecida.

Cree una cola para almacenar todas las URL de la frontera de rastreo y coloque el primer elemento como una URL semilla.

class MultiThreadedCrawler:
  
    def __init__(self, seed_url):
        self.seed_url = seed_url
        self.root_url = '{}://{}'.format(urlparse(self.seed_url).scheme,
                                         urlparse(self.seed_url).netloc)
        self.pool = ThreadPoolExecutor(max_workers=5)
        self.scraped_pages = set([])
        self.crawl_queue = Queue()
        self.crawl_queue.put(self.seed_url)

Paso 4: Cree un método llamado run_web_crawler(), para seguir agregando el enlace a la frontera y extrayendo la información, use un ciclo while infinito y muestre el nombre del proceso que se está ejecutando actualmente.

Obtenga la URL de la frontera de rastreo, para la búsqueda, asigne un tiempo de espera de 60 segundos y verifique si la URL actual ya se visitó o no. Si aún no lo ha visitado, formatee la URL actual y agréguela a scraped_pages configurado para almacenar en el historial de páginas visitadas y elija de un grupo de hilos y pase la página de raspado y la URL de destino.

def run_web_crawler(self):
    while True:
        try:
            print("\n Name of the current executing process: ",
                  multiprocessing.current_process().name, '\n')
            target_url = self.crawl_queue.get(timeout=60)
              
            if target_url not in self.scraped_pages:
                
                print("Scraping URL: {}".format(target_url))
                self.scraped_pages.add(target_url)
                job = self.pool.submit(self.scrape_page, target_url)
                job.add_done_callback(self.post_scrape_callback)
  
        except Empty:
            return
        except Exception as e:
            print(e)
            continue

Paso 5: Usando el método de protocolo de enlace, coloque la solicitud y establezca el tiempo predeterminado en 3 y el tiempo máximo en 30 y, una vez que la solicitud sea exitosa, devuelva el conjunto de resultados.

def scrape_page(self, url):
    try:
        res = requests.get(url, timeout=(3, 30))
        return res
    except requests.RequestException:
        return

Paso 6: Crea un método llamado scrape_info(). Y pase los datos de la página web a BeautifulSoap, lo que nos ayuda a organizar y formatear los datos web desordenados al corregir el HTML incorrecto y presentarnos en una estructura fácilmente transitable.

Usando el operador BeautifulSoup, extraiga todo el texto presente en el documento HTML.

def scrape_info(self, html):
    soup = BeautifulSoup(html, "html5lib")
    web_page_paragraph_contents = soup('p')
    text = ''
      
    for para in web_page_paragraph_contents:
        if not ('https:' in str(para.text)):
            text = text + str(para.text).strip()
    print('\n <-----Text Present in The WebPage is--->\n', text, '\n')
    return

Paso 7: Cree un método llamado parse links, utilizando el operador BeautifulSoup para extraer todas las etiquetas de anclaje presentes en el documento HTML. Soup.find_all(‘a’,href=True) devuelve una lista de elementos que contienen todas las etiquetas de anclaje presentes en la página web. Almacene todas las etiquetas en una lista llamada Anchor_Tags. Para cada etiqueta de anclaje presente en la lista Aachor_Tags, recupere el valor asociado con href en la etiqueta usando Link[‘href’]. Para cada URL recuperada, compruebe si es una URL absoluta o una URL relativa.

• URL relativa: URL sin URL raíz ni nombres de protocolo.
• URLs absolutas: URL con nombre de protocolo, URL raíz, nombre del documento.

Si es una URL relativa que usa el método urljoin, cámbiela a una URL absoluta usando la URL base y la URL relativa. Compruebe si la URL actual ya ha sido visitada o no. Si la URL aún no ha sido visitada, colóquela en la cola de rastreo.

def parse_links(self, html):
    soup = BeautifulSoup(html, 'html.parser')
    Anchor_Tags = soup.find_all('a', href=True)
      
    for link in Anchor_Tags:
        url = link['href']
          
        if url.startswith('/') or url.startswith(self.root_url):
            url = urljoin(self.root_url, url)
              
            if url not in self.scraped_pages:
                self.crawl_queue.put(url)

Paso 8: para extraer los enlaces, llame al método llamado parse_links() y pase el resultado. Para extraer el contenido, llame al método llamado scrape_info() y pase el resultado.

def post_scrape_callback(self, res):
    result = res.result()
      
    if result and result.status_code == 200:
        self.parse_links(result.text)
        self.scrape_info(result.text)

A continuación se muestra la implementación completa:

import multiprocessing
from bs4 import BeautifulSoup
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urljoin, urlparse
import requests
  
  
class MultiThreadedCrawler:
  
    def __init__(self, seed_url):
        self.seed_url = seed_url
        self.root_url = '{}://{}'.format(urlparse(self.seed_url).scheme,
                                         urlparse(self.seed_url).netloc)
        self.pool = ThreadPoolExecutor(max_workers=5)
        self.scraped_pages = set([])
        self.crawl_queue = Queue()
        self.crawl_queue.put(self.seed_url)
  
    def parse_links(self, html):
        soup = BeautifulSoup(html, 'html.parser')
        Anchor_Tags = soup.find_all('a', href=True)
        for link in Anchor_Tags:
            url = link['href']
            if url.startswith('/') or url.startswith(self.root_url):
                url = urljoin(self.root_url, url)
                if url not in self.scraped_pages:
                    self.crawl_queue.put(url)
  
    def scrape_info(self, html):
        soup = BeautifulSoup(html, "html5lib")
        web_page_paragraph_contents = soup('p')
        text = ''
        for para in web_page_paragraph_contents:
            if not ('https:' in str(para.text)):
                text = text + str(para.text).strip()
        print(f'\n <---Text Present in The WebPage is --->\n', text, '\n')
        return
  
    def post_scrape_callback(self, res):
        result = res.result()
        if result and result.status_code == 200:
            self.parse_links(result.text)
            self.scrape_info(result.text)
  
    def scrape_page(self, url):
        try:
            res = requests.get(url, timeout=(3, 30))
            return res
        except requests.RequestException:
            return
  
    def run_web_crawler(self):
        while True:
            try:
                print("\n Name of the current executing process: ",
                      multiprocessing.current_process().name, '\n')
                target_url = self.crawl_queue.get(timeout=60)
                if target_url not in self.scraped_pages:
                    print("Scraping URL: {}".format(target_url))
                    self.current_scraping_url = "{}".format(target_url)
                    self.scraped_pages.add(target_url)
                    job = self.pool.submit(self.scrape_page, target_url)
                    job.add_done_callback(self.post_scrape_callback)
  
            except Empty:
                return
            except Exception as e:
                print(e)
                continue
  
    def info(self):
        print('\n Seed URL is: ', self.seed_url, '\n')
        print('Scraped pages are: ', self.scraped_pages, '\n')
  
  
if __name__ == '__main__':
    cc = MultiThreadedCrawler("https://www.geeksforgeeks.org/")
    cc.run_web_crawler()
    cc.info()

Producción: