Creazione del web crawler - AWS Guida prescrittiva

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Creazione del web crawler

Come descritto nella Architettura sezione, l'applicazione viene eseguita in batch, uno per ogni azienda.

Acquisizione ed elaborazione del file robots.txt

Dopo aver preparato il set di dati, è necessario confermare se il dominio ha un file robots.txt. Per i web crawler e altri bot, il file robots.txt indica quali sezioni del sito Web possono visitare. Rispettare le istruzioni contenute in questo file è una best practice importante per la scansione etica dei siti web. Per ulteriori informazioni, consulta le migliori pratiche per i web crawler etici in questa guida.

Per acquisire ed elaborare il file robots.txt
  1. Se non l'hai già fatto, installa la requests libreria eseguendo il seguente comando in un terminale:

    pip install requests
  2. Esegui il seguente script. Lo script svolge le funzioni seguenti:

    • Definisce una check_robots_txt funzione che accetta un dominio come input.

    • Costruisce l'URL completo per il file robots.txt.

    • Invia una richiesta GET all'URL per il file robots.txt.

    • Se la richiesta ha esito positivo (codice di stato 200), esiste un file robots.txt.

    • Se la richiesta ha esito negativo o restituisce un codice di stato diverso, il file robots.txt non esiste o non è accessibile.

    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 False
    Nota

    Questo script gestisce le eccezioni per errori di rete o altri problemi.

  3. Se esiste un file robots.txt, usa il seguente script per scaricarlo:

    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 e
    Nota

    Questi script possono essere personalizzati o modificati in base al caso d'uso. Puoi anche combinare questi script.

Acquisizione ed elaborazione della mappa del sito

Successivamente, è necessario elaborare la mappa del sito. Puoi utilizzare la mappa del sito per concentrare la scansione su pagine importanti. Ciò migliora l'efficienza della scansione. Per ulteriori informazioni, consulta le migliori pratiche per i web crawler etici in questa guida.

Per acquisire ed elaborare la mappa del sito
  • Esegui il seguente script. Questo script definisce una check_and_download_sitemap funzione che:

    • Accetta un URL di base, un URL opzionale della mappa del sito da robots.txt e una stringa user-agent.

    • Controlla diverse potenziali posizioni della mappa del sito, inclusa quella di robots.txt (se fornita).

    • Tenta di scaricare la mappa del sito da ogni posizione.

    • Verifica che il contenuto scaricato sia in formato XML.

    • Chiama la parse_sitemap funzione per estrarre il URLs. Questa funzione:

      • Analizza il contenuto XML della mappa del sito.

      • Gestisce sia le normali Sitemap che i file di indice delle Sitemap.

      • Recupera in modo ricorsivo le mappe secondarie se viene rilevato un indice della mappa del sito.

    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("...")

Progettazione del crawler

Successivamente, si progetta il web crawler. Il crawler è progettato per seguire le migliori pratiche descritte in Le migliori pratiche per i web crawler etici questa guida. Questo EthicalCrawler corso illustra diversi principi chiave della scansione etica:

  • Recupero e analisi del file robots.txt: il crawler recupera il file robots.txt per il sito Web di destinazione.

  • Rispetto delle autorizzazioni di scansione: prima di eseguire la scansione di qualsiasi URL, il crawler verifica se le regole del file robots.txt consentono la scansione di quell'URL. Se un URL non è consentito, il crawler lo salta e passa all'URL successivo.

  • Rispetto del ritardo di scansione: il crawler verifica la presenza di una direttiva crawl-delay nel file robots.txt. Se ne viene specificata una, il crawler utilizza questo ritardo tra le richieste. Altrimenti, utilizza un ritardo predefinito.

  • Identificazione utente-agente: il crawler utilizza una stringa user-agent personalizzata per identificarsi nei siti Web. Se necessario, i proprietari di siti Web possono impostare regole specifiche per limitare o consentire il crawler.

  • Gestione degli errori e riduzione graduale: se il file robots.txt non può essere recuperato o analizzato, il crawler procede con regole predefinite conservative. Gestisce gli errori di rete e le risposte HTTP diverse da 200.

  • Scansione limitata: per evitare di sovraccaricare il server, esiste un limite al numero di pagine che possono essere scansionate.

Lo script seguente è uno pseudocodice che spiega come funziona il web crawler:

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.")
Creare un web crawler etico e avanzato che raccolga dati ESG
  1. Copia il seguente esempio di codice per il web crawler etico avanzato utilizzato in questo 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)
  2. Imposta i vari attributi, inclusi gli agenti utente, le raccolte vuote e gli URLs elenchi di archiviazione dei dati.

  3. Modifica le parole chiave e i criteri di pertinenza del is_relevant_to_sustainable_finance() metodo in base alle tue esigenze specifiche.

  4. Accertatevi che il file robots.txt consenta la scansione del sito Web e che stiate utilizzando il ritardo di scansione e lo user agent specificati nel file robots.txt.

  5. Valuta la possibilità di apportare le seguenti personalizzazioni allo script del crawler Web fornito, in base alle esigenze della tua organizzazione:

    • Implementate il fetch_sitemap() metodo per una più efficiente individuazione degli URL.

    • Migliora la registrazione e la gestione degli errori per l'uso in produzione.

    • Implementa un'analisi più sofisticata della pertinenza dei contenuti.

    • Aggiungi controlli di profondità e ampiezza per limitare l'ambito di scansione.