Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
Den Webcrawler erstellen
Wie im Architektur Abschnitt beschrieben, wird die Anwendung in Batches ausgeführt — einer für jedes Unternehmen.
Themen
Erfassen und Verarbeiten der Datei robots.txt
Nachdem Sie den Datensatz vorbereitet haben, müssen Sie überprüfen, ob die Domain über eine Datei robots.txt
Um die Datei robots.txt zu erfassen und zu verarbeiten
-
Falls Sie dies noch nicht getan haben, installieren Sie die
requestsBibliothek, indem Sie den folgenden Befehl in einem Terminal ausführen:pip install requests -
Führen Sie das folgende Skript aus. Dieses Skript führt folgende Aktionen aus:
-
Es definiert eine
check_robots_txtFunktion, die eine Domain als Eingabe verwendet. -
Es erstellt die vollständige URL für die Datei robots.txt.
-
Es sendet eine GET-Anfrage an die URL für die Datei robots.txt.
-
Wenn die Anfrage erfolgreich ist (Statuscode 200), ist eine Datei robots.txt vorhanden.
-
Wenn die Anfrage fehlschlägt oder einen anderen Statuscode zurückgibt, ist die Datei robots.txt nicht vorhanden oder es kann nicht darauf zugegriffen werden.
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 FalseAnmerkung
Dieses Skript behandelt Ausnahmen für Netzwerkfehler oder andere Probleme.
-
-
Wenn eine Datei robots.txt existiert, verwenden Sie das folgende Skript, um sie herunterzuladen:
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 eAnmerkung
Diese Skripts können je nach Anwendungsfall angepasst oder geändert werden. Sie können diese Skripte auch kombinieren.
Erfassung und Verarbeitung der Sitemap
Als Nächstes müssen Sie die Sitemap
Um die Sitemap zu erfassen und zu verarbeiten
-
Führen Sie das folgende Skript aus. Dieses Skript definiert eine
check_and_download_sitemapFunktion, die:-
Akzeptiert eine Basis-URL, eine optionale Sitemap-URL aus robots.txt und eine User-Agent-Zeichenfolge.
-
Überprüft mehrere potenzielle Sitemap-Speicherorte, einschließlich des Speicherorts von robots.txt (falls angegeben).
-
Versucht, die Sitemap von jedem Standort herunterzuladen.
-
Überprüft, ob der heruntergeladene Inhalt im XML-Format vorliegt.
-
Ruft die
parse_sitemapFunktion zum Extrahieren von auf URLs. Diese Funktion:-
Analysiert den XML-Inhalt der Sitemap.
-
Verarbeitet sowohl reguläre Sitemaps als auch Sitemap-Indexdateien.
-
Ruft rekursiv Unter-Sitemaps ab, wenn ein Sitemap-Index gefunden wird.
-
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("...") -
Den Crawler entwerfen
Als Nächstes entwerfen Sie den Webcrawler. Der Crawler ist so konzipiert, dass er den Bewährte Methoden für ethische Webcrawler in diesem Handbuch beschriebenen bewährten Methoden folgt. In diesem EthicalCrawler Kurs werden mehrere wichtige Prinzipien des ethischen Crawlings vorgestellt:
-
Die Datei robots.txt abrufen und analysieren — Der Crawler ruft die Datei robots.txt für die Zielwebsite ab.
-
Einhaltung der Crawling-Berechtigungen — Vor dem Crawlen einer URL überprüft der Crawler, ob die Regeln in der Datei robots.txt das Crawlen für diese URL zulassen. Wenn eine URL nicht erlaubt ist, überspringt der Crawler sie und wechselt zur nächsten URL.
-
Berücksichtigung der Crawlverzögerung — Der Crawler sucht in der Datei robots.txt nach einer Crawl-Delay-Direktive. Wenn eine angegeben ist, verwendet der Crawler diese Verzögerung zwischen Anfragen. Andernfalls verwendet er eine Standardverzögerung.
-
User-Agent-Identifikation — Der Crawler verwendet eine benutzerdefinierte User-Agent-Zeichenfolge, um sich gegenüber Websites zu identifizieren. Bei Bedarf können Webseitenbesitzer spezifische Regeln festlegen, um deinen Crawler einzuschränken oder zuzulassen.
-
Fehlerbehandlung und ordentliche Degradierung — Wenn die Datei robots.txt nicht abgerufen oder analysiert werden kann, verwendet der Crawler konservative Standardregeln. Er behandelt Netzwerkfehler und HTTP-Antworten, die nicht 200 sind.
-
Eingeschränktes Crawlen — Um eine Überlastung des Servers zu vermeiden, gibt es ein Limit für die Anzahl der Seiten, die gecrawlt werden können.
Das folgende Skript ist Pseudocode, der erklärt, wie der Webcrawler funktioniert:
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.")
Um einen fortschrittlichen, ethischen Webcrawler zu entwickeln, der ESG-Daten sammelt
-
Kopieren Sie das folgende Codebeispiel für den fortschrittlichen ethischen Webcrawler, der in diesem System verwendet wird:
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) -
Richten Sie die verschiedenen Attribute ein, darunter Benutzeragenten, leere Sammlungen für URLs und Datenspeicherlisten.
-
Passen Sie die Schlüsselwörter und Relevanzkriterien in der
is_relevant_to_sustainable_finance()Methode an Ihre spezifischen Bedürfnisse an. -
Stellen Sie sicher, dass die Datei robots.txt das Crawlen der Website zulässt und dass Sie die in der Datei robots.txt angegebene Crawling-Verzögerung und den Benutzeragenten verwenden.
-
Erwägen Sie, das bereitgestellte Webcrawler-Skript nach Bedarf für Ihre Organisation wie folgt anzupassen:
-
Implementieren Sie die
fetch_sitemap()Methode für eine effizientere URL-Erkennung. -
Verbessern Sie die Fehlerprotokollierung und -behandlung für den Produktionseinsatz.
-
Implementieren Sie eine ausgefeiltere Analyse der Relevanz von Inhalten.
-
Fügen Sie Kontrollen für Tiefe und Breite hinzu, um den Umfang des Crawlings einzuschränken.
-
