翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# ウェブクローラーの構築
[アーキテクチャ](architecture.md) 「」セクションで説明されているように、アプリケーションは会社ごとに 1 つずつバッチで実行されます。
**Topics**
+ [robots.txt ファイルのキャプチャと処理](#building-crawler-robots-txt)
+ [サイトマップのキャプチャと処理](#building-crawler-sitemap)
+ [クローラの設計](#building-crawler-design)
## robots.txt ファイルのキャプチャと処理
データセットを準備したら、ドメインに [robots.txt](https://en.wikipedia.org/wiki/Robots.txt) ファイルがあるかどうかを確認する必要があります。ウェブクローラーやその他のボットの場合、robots.txt ファイルはウェブサイトのどのセクションにアクセスできるかを示します。このファイルの指示に従うことは、ウェブサイトを倫理的にクロールするための重要なベストプラクティスです。詳細については、このガイドの[「倫理的ウェブクローラーのベストプラクティス](best-practices.md)」を参照してください。
**robots.txt ファイルをキャプチャして処理するには**
1. まだインストールしていない場合は、ターミナルで次のコマンドを実行して`requests`ライブラリをインストールします。
```
pip install requests
```
1. 次のスクリプトを実行します。このスクリプトは以下の処理を実行します:
+ ドメインを入力として受け取る`check_robots_txt`関数を定義します。
+ robots.txt ファイルの完全な URL を作成します。
+ robots.txt ファイルの URL に GET リクエストを送信します。
+ リクエストが成功した場合 (ステータスコード 200)、robots.txt ファイルがあります。
+ リクエストが失敗するか、別のステータスコードを返す場合、robots.txt ファイルが存在しないか、アクセスできません。
```
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
```
**注記**
このスクリプトは、ネットワークエラーやその他の問題の例外を処理します。
1. robots.txt ファイルが存在する場合は、次のスクリプトを使用してダウンロードします。
```
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
```
**注記**
これらのスクリプトは、ユースケースに応じてカスタマイズまたは変更できます。これらのスクリプトを組み合わせることもできます。
## サイトマップのキャプチャと処理
次に、[サイトマップ](https://en.wikipedia.org/wiki/Site_map)を処理する必要があります。サイトマップを使用して、重要なページのクロールに集中できます。これにより、クローリングの効率が向上します。詳細については、このガイドの[「倫理的ウェブクローラーのベストプラクティス](best-practices.md)」を参照してください。
**サイトマップをキャプチャして処理するには**
+ 次のスクリプトを実行します。このスクリプトは、以下の`check_and_download_sitemap`関数を定義します。
+ ベース URL、robots.txt からのオプションのサイトマップ URL、およびユーザーエージェント文字列を受け入れます。
+ robots.txt (提供されている場合) からのサイトマップを含む、複数の潜在的なサイトマップの場所をチェックします。
+ 各場所からサイトマップをダウンロードしようとします。
+ ダウンロードしたコンテンツが XML 形式であることを確認します。
+ `parse_sitemap` 関数を呼び出して URLs。この関数:
+ サイトマップの XML コンテンツを解析します。
+ 通常のサイトマップとサイトマップインデックスファイルの両方を処理します。
+ サイトマップインデックスが発生した場合、サブサイトマップを再帰的に取得します。
```
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("...")
```
## クローラの設計
次に、ウェブクローラーを設計します。クローラは、このガイド[倫理的ウェブクローラーのベストプラクティス](best-practices.md)で説明されているベストプラクティスに従うように設計されています。この`EthicalCrawler`クラスは、倫理的クロールのいくつかの主要な原則を示しています。
+ **robots.txt ファイルの取得と解析** — クローラはターゲットウェブサイトの robots.txt ファイルを取得します。
+ **クローリング許可の尊重** – URL をクローリングする前に、クローラは robots.txt ファイルのルールでその URL のクローリングが許可されているかどうかを確認します。URL が許可されていない場合、クローラはその URL をスキップして次の URL に移動します。
+ **クロール遅延の尊重** – クローラーは robots.txt ファイル内のクロール遅延ディレクティブをチェックします。指定した場合、クローラはリクエスト間でこの遅延を使用します。それ以外の場合は、デフォルトの遅延を使用します。
+ **ユーザーエージェント識別** – クローラはカスタムユーザーエージェント文字列を使用して、ウェブサイトに対して自身を識別します。必要に応じて、ウェブサイトの所有者は、クローラを制限または許可する特定のルールを設定できます。
+ **エラー処理と正常な低下** — robots.txt ファイルをフェッチまたは解析できない場合、クローラは保守的なデフォルトルールに進みます。ネットワークエラーと 200 以外の HTTP レスポンスを処理します。
+ **制限付きクローリング** – サーバーの負担を避けるために、クローリングできるページ数に制限があります。
次のスクリプトは、ウェブクローラーの仕組みを説明する擬似コードです。
```
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.")
```
**ESG データを収集する高度な倫理的ウェブクローラーを構築するには**
1. このシステムで使用される高度な倫理的ウェブクローラーの次のコードサンプルをコピーします。
```
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)
```
1. ユーザーエージェント、URLs の空のコレクション、データストレージリストなど、さまざまな属性を設定します。
1. `is_relevant_to_sustainable_finance()` メソッドのキーワードと関連性基準を、特定のニーズに合わせて調整します。
1. robots.txt ファイルがウェブサイトのクロールを許可し、robots.txt ファイルで指定されたクロール遅延とユーザーエージェントを使用していることを確認します。
1. 組織で必要に応じて、提供されたウェブクローラースクリプトに次のカスタマイズを行うことを検討してください。
+ より効率的な URL 検出のために `fetch_sitemap()`メソッドを実装します。
+ 本番稼働用のエラーログ記録と処理を強化します。
+ より高度なコンテンツ関連性分析を実装します。
+ 深さと幅のコントロールを追加して、クローリング範囲を制限します。