robots.txt ファイルのキャプチャと処理サイトマップのキャプチャと処理クローラの設計

ウェブクローラーの構築

アーキテクチャ「」セクションで説明されているように、アプリケーションは会社ごとに 1 つずつバッチで実行されます。

トピック

robots.txt ファイルのキャプチャと処理
サイトマップのキャプチャと処理
クローラの設計

robots.txt ファイルのキャプチャと処理

データセットを準備したら、ドメインに robots.txt ファイルがあるかどうかを確認する必要があります。ウェブクローラーやその他のボットの場合、robots.txt ファイルには、アクセスが許可されているウェブサイトのセクションが表示されます。このファイルの指示に従うことは、ウェブサイトを倫理的にクロールするための重要なベストプラクティスです。詳細については、このガイドの「倫理的ウェブクローラーのベストプラクティス」を参照してください。

robots.txt ファイルをキャプチャして処理するには

まだインストールしていない場合は、ターミナルで次のコマンドを実行してrequestsライブラリをインストールします。
```
pip install requests
```

次のスクリプトを実行します。このスクリプトは以下の処理を実行します:

ドメインを入力として受け取る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

注記

このスクリプトは、ネットワークエラーやその他の問題の例外を処理します。

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

注記

これらのスクリプトは、ユースケースに応じてカスタマイズまたは変更できます。これらのスクリプトを組み合わせることもできます。

サイトマップのキャプチャと処理

次に、サイトマップを処理する必要があります。サイトマップを使用して、重要なページのクロールに集中できます。これにより、クローリング効率が向上します。詳細については、このガイドの「倫理的ウェブクローラーのベストプラクティス」を参照してください。

サイトマップをキャプチャして処理するには

次のスクリプトを実行します。このスクリプトは、以下の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("...")

クローラの設計

次に、ウェブクローラーを設計します。クローラは、このガイド倫理的ウェブクローラーのベストプラクティスで説明されているベストプラクティスに従うように設計されています。この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 データを収集する高度な倫理的ウェブクローラーを構築するには

このシステムで使用される高度な倫理的ウェブクローラーの次のコードサンプルをコピーします。


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)

ユーザーエージェント、URLs の空のコレクション、データストレージリストなど、さまざまな属性を設定します。
is_relevant_to_sustainable_finance() メソッドのキーワードと関連性基準を、特定のニーズに合わせて調整します。
robots.txt ファイルがウェブサイトのクロールを許可し、robots.txt ファイルで指定されたクロール遅延とユーザーエージェントを使用していることを確認します。
組織で必要に応じて、提供されたウェブクローラースクリプトに次のカスタマイズを行うことを検討してください。
- より効率的な URL 検出のために fetch_sitemap()メソッドを実装します。
- 本番環境でのエラーのログ記録と処理を強化します。
- より高度なコンテンツ関連性分析を実装します。
- 深さと幅のコントロールを追加して、クローリング範囲を制限します。

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

データセットの準備

AWS インフラストラクチャの構築