Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
Construir el rastreador web
Como se describe en la Arquitectura sección, la aplicación se ejecuta en lotes, uno para cada empresa.
Temas
Capturar y procesar el archivo robots.txt
Después de preparar el conjunto de datos, debe confirmar si el dominio tiene un archivo robots.txt
Para capturar y procesar el archivo robots.txt
-
Si aún no lo ha hecho, instale la
requestsbiblioteca ejecutando el siguiente comando en una terminal:pip install requests -
Ejecute el siguiente script. Este script hace lo siguiente:
-
Define una
check_robots_txtfunción que toma un dominio como entrada. -
Construye la URL completa del archivo robots.txt.
-
Envía una solicitud GET a la URL del archivo robots.txt.
-
Si la solicitud se realiza correctamente (código de estado 200), existe un archivo robots.txt.
-
Si la solicitud falla o devuelve un código de estado diferente, significa que el archivo robots.txt no existe o no está accesible.
import requests from urllib.parse import urljoin def check_robots_txt(domain): # Ensure the domain starts with a protocol if not domain.startswith(('http://', 'https://')): domain = 'https://' + domain # Construct the full URL for robots.txt robots_url = urljoin(domain, '/robots.txt') try: # Send a GET request to the robots.txt URL response = requests.get(robots_url, timeout=5) # Check if the request was successful (status code 200) if response.status_code == 200: print(f"robots.txt found at {robots_url}") return True else: print(f"No robots.txt found at {robots_url} (Status code: {response.status_code})") return False except requests.RequestException as e: print(f"Error checking {robots_url}: {e}") return Falsenota
Este script gestiona las excepciones por errores de red u otros problemas.
-
-
Si existe un archivo robots.txt, utilice el siguiente script para descargarlo:
import requests def download(self, url): response = requests.get(url, headers=self.headers, timeout=5) response.raise_for_status() # Raise an exception for non-2xx responses return response.text def download_robots_txt(self): # Append '/robots.txt' to the URL to get the robots.txt file's URL robots_url = self.url.rstrip('/') + '/robots.txt' try: response = download(robots_url) return response except requests.exceptions.RequestException as e: print(f"Error downloading robots.txt: {e}, \nGenerating sitemap using combinations...") return enota
Estos scripts se pueden personalizar o modificar según su caso de uso. También puede combinar estos scripts.
Capturar y procesar el mapa del sitio
A continuación, debe procesar el mapa del sitio
Para capturar y procesar el mapa del sitio
-
Ejecute el siguiente script. Este script define una
check_and_download_sitemapfunción que:-
Acepta una URL base, una URL de mapa del sitio opcional de robots.txt y una cadena de agente de usuario.
-
Comprueba varias ubicaciones posibles del mapa del sitio, incluida la de robots.txt (si se proporciona).
-
Intenta descargar el mapa del sitio desde cada ubicación.
-
Comprueba que el contenido descargado esté en formato XML.
-
Llama a la
parse_sitemapfunción para extraer el URLs. Esta función:-
Analiza el contenido XML del mapa del sitio.
-
Gestiona tanto los mapas de sitio normales como los archivos de índice de los mismos.
-
Busca de forma recursiva los submapas del sitio si encuentra un índice de mapa del sitio.
-
import requests from urllib.parse import urljoin import xml.etree.ElementTree as ET def check_and_download_sitemap(base_url, robots_sitemap_url=None, user_agent='SitemapBot/1.0'): headers = {'User-Agent': user_agent} sitemap_locations = [robots_sitemap_url, urljoin(base_url, '/sitemap.xml'), urljoin(base_url, '/sitemap_index.xml'), urljoin(base_url, '/sitemap/'), urljoin(base_url, '/sitemap/sitemap.xml')] for sitemap_url in sitemap_locations: if not sitemap_url: continue print(f"Checking for sitemap at: {sitemap_url}") try: response = requests.get(sitemap_url, headers=headers, timeout=10) if response.status_code == 200: content_type = response.headers.get('Content-Type', '') if 'xml' in content_type: print(f"Successfully downloaded sitemap from {sitemap_url}") return parse_sitemap(response.text) else: print(f"Found content at {sitemap_url}, but it's not XML. Content-Type: {content_type}") except requests.RequestException as e: print(f"Error downloading sitemap from {sitemap_url}: {e}") print("No sitemap found.") return [] def parse_sitemap(sitemap_content): urls = [] try: root = ET.fromstring(sitemap_content) # Handle both sitemap and sitemapindex for loc in root.findall('.//{http://www.sitemaps.org/schemas/sitemap/0.9}loc'): urls.append(loc.text) # If it's a sitemap index, recursively fetch each sitemap if root.tag.endswith('sitemapindex'): all_urls = [] for url in urls: print(f"Fetching sub-sitemap: {url}") sub_sitemap_urls = check_and_download_sitemap(url) all_urls.extend(sub_sitemap_urls) return all_urls except ET.ParseError as e: print(f"Error parsing sitemap XML: {e}") return urls if __name__ == "__main__": base_url = input("Enter the base URL of the website: ") robots_sitemap_url = input("Enter the sitemap URL from robots.txt (or press Enter if none): ").strip() or None urls = check_and_download_sitemap(base_url, robots_sitemap_url) print(f"Found {len(urls)} URLs in sitemap:") for url in urls[:5]: # Print first 5 URLs as an example print(url) if len(urls) > 5: print("...") -
Diseñar el rastreador
A continuación, diseñe el rastreador web. El rastreador está diseñado para seguir las prácticas recomendadas que se describen Mejores prácticas para rastreadores web éticos en esta guía. En esta EthicalCrawler clase se muestran varios principios clave del rastreo ético:
-
Obtener y analizar el archivo robots.txt: el rastreador busca el archivo robots.txt del sitio web de destino.
-
Respetar los permisos de rastreo: antes de rastrear cualquier URL, el rastreador comprueba si las reglas del archivo robots.txt permiten rastrear esa URL. Si una URL no está permitida, el rastreador la omite y pasa a la siguiente URL.
-
Respetar el retraso de rastreo: el rastreador comprueba si hay una directiva de retraso de rastreo en el archivo robots.txt. Si se especifica alguna, el rastreador utiliza este retraso entre las solicitudes. De lo contrario, utiliza un retraso predeterminado.
-
Identificación del agente de usuario: el rastreador utiliza una cadena de agente de usuario personalizada para identificarse en los sitios web. Si es necesario, los propietarios de los sitios web pueden establecer reglas específicas para restringir o permitir el rastreo.
-
Gestión de errores y degradación correcta: si el archivo robots.txt no se puede recuperar ni analizar, el rastreador sigue las reglas predeterminadas conservadoras. Gestiona los errores de red y las respuestas HTTP distintas de las 200.
-
Rastreo limitado: para evitar sobrecargar el servidor, existe un límite en cuanto al número de páginas que se pueden rastrear.
El siguiente script es un pseudocódigo que explica cómo funciona el rastreador web:
import requests from urllib.parse import urljoin, urlparse import time class EthicalCrawler: def __init__(self, start_url, user_agent='EthicalBot/1.0'): self.start_url = start_url self.user_agent = user_agent self.domain = urlparse(start_url).netloc self.robots_parser = None self.crawl_delay = 1 # Default delay in seconds def can_fetch(self, url): if self.robots_parser: return self.robots_parser.allowed(url, self.user_agent) return True # If no robots.txt, assume allowed but crawl conservatively def get_crawl_delay(self): if self.robots_parser: delay = self.robots_parser.agent(self.user_agent).delay if delay is not None: self.crawl_delay = delay print(f"Using crawl delay of {self.crawl_delay} seconds") def crawl(self, max_pages=10): self.get_crawl_delay() pages_crawled = 0 urls_to_crawl = [self.start_url] while urls_to_crawl and pages_crawled < max_pages: url = urls_to_crawl.pop(0) if not self.can_fetch(url): print(f"robots.txt disallows crawling: {url}") continue try: response = requests.get(url, headers={'User-Agent': self.user_agent}) if response.status_code == 200: print(f"Successfully crawled: {url}") # Here you would typically parse the content, extract links, etc. # For this example, we'll just increment the counter pages_crawled += 1 else: print(f"Failed to crawl {url}: HTTP {response.status_code}") except Exception as e: print(f"Error crawling {url}: {e}") # Respect the crawl delay time.sleep(self.crawl_delay) print(f"Crawling complete. Crawled {pages_crawled} pages.")
Para crear un rastreador web avanzado y ético que recopile datos de ESG
-
Copie el siguiente ejemplo de código para el rastreador web ético avanzado que se utiliza en este sistema:
import requests from urllib.parse import urljoin, urlparse import time from collections import deque import random from bs4 import BeautifulSoup import re import csv import os class EnhancedESGCrawler: def __init__(self, start_url): self.start_url = start_url self.domain = urlparse(start_url).netloc self.desktop_user_agent = 'ESGEthicalBot/1.0' self.mobile_user_agent = 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1' self.robots_parser = None self.crawl_delay = None self.urls_to_crawl = deque() self.crawled_urls = set() self.max_retries = 2 self.session = requests.Session() self.esg_data = [] self.news_links = [] self.pdf_links = [] def setup(self): self.fetch_robots_txt() # Provided in Previous Snippet self.fetch_sitemap() # Provided in Previous Snippet def can_fetch(self, url, user_agent): if self.robots_parser: return self.robots_parser.allowed(url, user_agent) return True def delay(self): if self.crawl_delay is not None: time.sleep(self.crawl_delay) else: time.sleep(random.uniform(1, 3)) def get_headers(self, user_agent): return {'User-Agent': user_agent, 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate, br', 'DNT': '1', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1'} def extract_esg_data(self, url, html_content): soup = BeautifulSoup(html_content, 'html.parser') esg_data = { 'url': url, 'environmental': self.extract_environmental_data(soup), 'social': self.extract_social_data(soup), 'governance': self.extract_governance_data(soup) } self.esg_data.append(esg_data) # Extract news links and PDFs self.extract_news_links(soup, url) self.extract_pdf_links(soup, url) def extract_environmental_data(self, soup): keywords = ['carbon footprint', 'emissions', 'renewable energy', 'waste management', 'climate change'] return self.extract_keyword_data(soup, keywords) def extract_social_data(self, soup): keywords = ['diversity', 'inclusion', 'human rights', 'labor practices', 'community engagement'] return self.extract_keyword_data(soup, keywords) def extract_governance_data(self, soup): keywords = ['board structure', 'executive compensation', 'shareholder rights', 'ethics', 'transparency'] return self.extract_keyword_data(soup, keywords) def extract_keyword_data(self, soup, keywords): text = soup.get_text().lower() return {keyword: len(re.findall(r'\b' + re.escape(keyword) + r'\b', text)) for keyword in keywords} def extract_news_links(self, soup, base_url): news_keywords = ['news', 'press release', 'article', 'blog', 'sustainability'] for a in soup.find_all('a', href=True): if any(keyword in a.text.lower() for keyword in news_keywords): full_url = urljoin(base_url, a['href']) if full_url not in self.news_links: self.news_links.append({'url': full_url, 'text': a.text.strip()}) def extract_pdf_links(self, soup, base_url): for a in soup.find_all('a', href=True): if a['href'].lower().endswith('.pdf'): full_url = urljoin(base_url, a['href']) if full_url not in self.pdf_links: self.pdf_links.append({'url': full_url, 'text': a.text.strip()}) def is_relevant_to_sustainable_finance(self, text): keywords = ['sustainable finance', 'esg', 'green bond', 'social impact', 'environmental impact', 'climate risk', 'sustainability report', 'corporate responsibility'] return any(keyword in text.lower() for keyword in keywords) def attempt_crawl(self, url, user_agent): for _ in range(self.max_retries): try: response = self.session.get(url, headers=self.get_headers(user_agent), timeout=10) if response.status_code == 200: print(f"Successfully crawled: {url}") if response.headers.get('Content-Type', '').startswith('text/html'): self.extract_esg_data(url, response.text) elif response.headers.get('Content-Type', '').startswith('application/pdf'): self.save_pdf(url, response.content) return True else: print(f"Failed to crawl {url}: HTTP {response.status_code}") except requests.RequestException as e: print(f"Error crawling {url} with {user_agent}: {e}") self.delay() return False def crawl_url(self, url): if not self.can_fetch(url, self.desktop_user_agent): print(f"Robots.txt disallows desktop user agent: {url}") if self.can_fetch(url, self.mobile_user_agent): print(f"Attempting with mobile user agent: {url}") return self.attempt_crawl(url, self.mobile_user_agent) else: print(f"Robots.txt disallows both user agents: {url}") return False return self.attempt_crawl(url, self.desktop_user_agent) def crawl(self, max_pages=100): self.setup() if not self.urls_to_crawl: self.urls_to_crawl.append(self.start_url) pages_crawled = 0 while self.urls_to_crawl and pages_crawled < max_pages: url = self.urls_to_crawl.popleft() if url not in self.crawled_urls: if self.crawl_url(url): pages_crawled += 1 self.crawled_urls.add(url) self.delay() print(f"Crawling complete. Successfully crawled {pages_crawled} pages.") self.save_esg_data() self.save_news_links() self.save_pdf_links() def save_esg_data(self): with open('esg_data.csv', 'w', newline='', encoding='utf-8') as file: writer = csv.DictWriter(file, fieldnames=['url', 'environmental', 'social', 'governance']) writer.writeheader() for data in self.esg_data: writer.writerow({ 'url': data['url'], 'environmental': ', '.join([f"{k}: {v}" for k, v in data['environmental'].items()]), 'social': ', '.join([f"{k}: {v}" for k, v in data['social'].items()]), 'governance': ', '.join([f"{k}: {v}" for k, v in data['governance'].items()]) }) print("ESG data saved to esg_data.csv") def save_news_links(self): with open('news_links.csv', 'w', newline='', encoding='utf-8') as file: writer = csv.DictWriter(file, fieldnames=['url', 'text', 'relevant']) writer.writeheader() for news in self.news_links: writer.writerow({ 'url': news['url'], 'text': news['text'], 'relevant': self.is_relevant_to_sustainable_finance(news['text']) }) print("News links saved to news_links.csv") def save_pdf_links(self): # Code for saving PDF in S3 or filesystem def save_pdf(self, url, content): # Code for saving PDF in S3 or filesystem # Example usage if __name__ == "__main__": start_url = input("Enter the starting URL to crawl for ESG data and news: ") crawler = EnhancedESGCrawler(start_url) crawler.crawl(max_pages=50) -
Configure los distintos atributos, incluidos los agentes de usuario, las colecciones vacías y las URLs listas de almacenamiento de datos.
-
Ajusta las palabras clave y los criterios de relevancia del
is_relevant_to_sustainable_finance()método para que se ajusten a tus necesidades específicas. -
Asegúrese de que el archivo robots.txt permite rastrear el sitio web y de que utiliza el retardo de rastreo y el agente de usuario especificados en el archivo robots.txt.
-
Considere la posibilidad de realizar las siguientes personalizaciones en el script de rastreo web proporcionado, según sea necesario para su organización:
-
Implemente el
fetch_sitemap()método para descubrir las URL de forma más eficiente. -
Mejore el registro y la gestión de errores para su uso en producción.
-
Implemente un análisis de relevancia del contenido más sofisticado.
-
Añada controles de profundidad y amplitud para limitar el alcance del rastreo.
-
