Menangkap dan memproses file robots.txt Menangkap dan memproses peta situs Merancang crawler

Membangun web crawler

Seperti yang dijelaskan di Arsitektur bagian ini, aplikasi berjalan dalam batch — satu untuk setiap perusahaan.

Topik

Menangkap dan memproses file robots.txt
Menangkap dan memproses peta situs
Merancang crawler

Menangkap dan memproses file robots.txt

Setelah menyiapkan kumpulan data, Anda perlu mengonfirmasi apakah domain tersebut memiliki file robots.txt. Untuk perayap web dan bot lainnya, file robots.txt menunjukkan bagian situs web mana yang diizinkan untuk dikunjungi. Menghormati instruksi dalam file ini adalah praktik terbaik yang penting untuk merayapi situs web secara etis. Untuk informasi selengkapnya, lihat Praktik terbaik untuk perayap web etis dalam panduan ini.

Untuk menangkap dan memproses file robots.txt

Jika Anda belum melakukannya, instal requests pustaka dengan menjalankan perintah berikut di terminal:
```
pip install requests
```

Jalankan skrip berikut. Skrip ini melakukan hal berikut:

Ini mendefinisikan check_robots_txt fungsi yang mengambil domain sebagai input.
Ini membangun URL lengkap untuk file robots.txt.
Ini mengirimkan permintaan GET ke URL untuk file robots.txt.
Jika permintaan berhasil (kode status 200), maka file robots.txt ada.
Jika permintaan gagal atau mengembalikan kode status yang berbeda, maka file robots.txt tidak ada atau tidak dapat diakses.


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

catatan

Skrip ini menangani pengecualian untuk kesalahan jaringan atau masalah lainnya.

Jika file robots.txt ada, gunakan skrip berikut untuk mengunduhnya:


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

catatan

Skrip ini dapat disesuaikan atau dimodifikasi sesuai dengan kasus penggunaan Anda. Anda juga dapat menggabungkan skrip ini.

Menangkap dan memproses peta situs

Selanjutnya, Anda perlu memproses peta situs. Anda dapat menggunakan peta situs untuk memfokuskan crawling pada halaman penting. Ini meningkatkan efisiensi perayapan. Untuk informasi selengkapnya, lihat Praktik terbaik untuk perayap web etis dalam panduan ini.

Untuk menangkap dan memproses peta situs

Jalankan skrip berikut. Script ini mendefinisikan check_and_download_sitemap fungsi yang:

Menerima URL dasar, URL peta situs opsional dari robots.txt, dan string user-agent.
Memeriksa beberapa lokasi peta situs potensial, termasuk yang dari robots.txt (jika disediakan).
Mencoba mengunduh peta situs dari setiap lokasi.
Memverifikasi bahwa konten yang diunduh dalam format XHTML.
Memanggil parse_sitemap fungsi untuk mengekstrak file URLs. Fungsi ini:
- Mem-parsing konten XHTML dari sitemap.
- Menangani peta situs biasa dan file indeks peta situs.
- Mengambil sub-peta situs secara rekursif jika indeks peta situs ditemukan.


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

Merancang crawler

Selanjutnya, Anda mendesain web crawler. Crawler dirancang untuk mengikuti praktik terbaik yang dijelaskan Praktik terbaik untuk perayap web etis dalam panduan ini. EthicalCrawlerKelas ini menunjukkan beberapa prinsip kunci merangkak etis:

Mengambil dan mengurai file robots.txt - Crawler mengambil file robots.txt untuk situs web target.
Menghormati izin perayapan — Sebelum merayapi URL apa pun, crawler memeriksa apakah aturan dalam file robots.txt memungkinkan crawling untuk URL tersebut. Jika URL tidak diizinkan, maka crawler melewatinya dan pindah ke URL berikutnya.
Menghormati penundaan crawl - Crawler memeriksa direktif penundaan perayapan di file robots.txt. Jika satu ditentukan, crawler menggunakan penundaan ini di antara permintaan. Jika tidak, ia menggunakan penundaan default.
Identifikasi agen pengguna - Crawler menggunakan string agen pengguna khusus untuk mengidentifikasi dirinya ke situs web. Jika diperlukan, pemilik situs web dapat menetapkan aturan khusus untuk membatasi atau mengizinkan crawler Anda.
Penanganan kesalahan dan degradasi anggun - Jika file robots.txt tidak dapat diambil atau diurai, crawler melanjutkan dengan aturan default konservatif. Ini menangani kesalahan jaringan dan respons HTTP non-200.
Perayapan terbatas — Untuk menghindari kewalahan server, ada batasan berapa banyak halaman yang dapat dirayapi.

Skrip berikut adalah pseudocode yang menjelaskan cara kerja crawler 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.")

Untuk membangun perayap web canggih dan etis yang mengumpulkan data ESG

Salin contoh kode berikut untuk crawler web etis tingkat lanjut yang digunakan dalam sistem ini:


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)

Siapkan berbagai atribut, termasuk agen pengguna, koleksi kosong untuk URLs, dan daftar penyimpanan data.
Sesuaikan kata kunci dan kriteria relevansi dalam is_relevant_to_sustainable_finance() metode agar sesuai dengan kebutuhan spesifik Anda.
Pastikan file robots.txt mengizinkan perayapan situs web dan Anda menggunakan penundaan perayapan dan agen pengguna yang ditentukan dalam file robots.txt.
Pertimbangkan untuk membuat penyesuaian berikut ke skrip perayap web yang disediakan, sesuai kebutuhan untuk organisasi Anda:
- Menerapkan fetch_sitemap() metode untuk penemuan URL yang lebih efisien.
- Tingkatkan pencatatan kesalahan dan penanganan untuk penggunaan produksi.
- Menerapkan analisis relevansi konten yang lebih canggih.
- Tambahkan kontrol kedalaman dan lebar untuk membatasi cakupan perayapan.

Awas Javascript dinonaktifkan atau tidak tersedia di browser Anda.

Untuk menggunakan Dokumentasi AWS, Javascript harus diaktifkan. Lihat halaman Bantuan browser Anda untuk petunjuk.

Konvensi Dokumen

Mempersiapkan dataset

Membangun AWS infrastruktur