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.
Python3
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().
Python3
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.
Python3
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.
Python3
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.
Python3
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.
Python3
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.
Python3
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.
Python3
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:
Python3
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:
Publicación traducida automáticamente
Artículo escrito por vinayedula y traducido por Barcelona Geeks. The original can be accessed here. Licence: CCBY-SA