Capturando e processando o arquivo robots.txt Capturando e processando o mapa do site Projetando o rastreador

Construindo o rastreador da web

Conforme descrito na Arquitetura seção, o aplicativo é executado em lotes — um para cada empresa.

Tópicos

Capturando e processando o arquivo robots.txt
Capturando e processando o mapa do site
Projetando o rastreador

Capturando e processando o arquivo robots.txt

Depois de preparar o conjunto de dados, você precisa confirmar se o domínio tem um arquivo robots.txt. Para rastreadores da web e outros bots, o arquivo robots.txt indica quais seções do site eles podem visitar. Cumprir as instruções desse arquivo é uma prática recomendada importante para rastrear sites de forma ética. Para obter mais informações, consulte Práticas recomendadas para rastreadores da Web éticos neste guia.

Para capturar e processar o arquivo robots.txt

Se você ainda não tiver feito isso, instale a requests biblioteca executando o seguinte comando em um terminal:
```
pip install requests
```

Consulte o script a seguir. Este script faz o seguinte:

Ele define uma check_robots_txt função que usa um domínio como entrada.
Ele constrói o URL completo para o arquivo robots.txt.
Ele envia uma solicitação GET para a URL do arquivo robots.txt.
Se a solicitação for bem-sucedida (código de status 200), existirá um arquivo robots.txt.
Se a solicitação falhar ou retornar um código de status diferente, o arquivo robots.txt não existe ou não está acessível.


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

Esse script trata exceções para erros de rede ou outros problemas.

Se existir um arquivo robots.txt, use o script a seguir para baixá-lo:


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

Esses scripts podem ser personalizados ou modificados de acordo com seu caso de uso. Você também pode combinar esses scripts.

Capturando e processando o mapa do site

Em seguida, você precisa processar o mapa do site. Você pode usar o mapa do site para focar no rastreamento de páginas importantes. Isso melhora a eficiência do rastreamento. Para obter mais informações, consulte Práticas recomendadas para rastreadores da Web éticos neste guia.

Para capturar e processar o mapa do site

Consulte o script a seguir. Esse script define uma check_and_download_sitemap função que:

Aceita um URL base, um URL opcional do sitemap do robots.txt e uma string de agente de usuário.
Verifica várias localizações potenciais do sitemap, incluindo a de robots.txt (se fornecida).
Tenta baixar o mapa do site de cada local.
Verifica se o conteúdo baixado está no formato XML.
Chama a parse_sitemap função para extrair URLs o. Esta função:
- Analisa o conteúdo XML do mapa do site.
- Lida com sitemaps regulares e arquivos de índice de sitemaps.
- Busca recursivamente subsitemaps se um índice de sitemap for encontrado.


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

Projetando o rastreador

Em seguida, você cria o rastreador da web. O rastreador foi projetado para seguir as melhores práticas descritas Práticas recomendadas para rastreadores da web éticos neste guia. Esta EthicalCrawler aula demonstra vários princípios fundamentais do rastreamento ético:

Buscando e analisando o arquivo robots.txt — O rastreador busca o arquivo robots.txt para o site de destino.
Respeitando as permissões de rastreamento — Antes de rastrear qualquer URL, o rastreador verifica se as regras no arquivo robots.txt permitem o rastreamento desse URL. Se um URL não for permitido, o rastreador o ignorará e passará para o próximo URL.
Honrando o atraso de rastreamento — O rastreador verifica se há uma diretiva de atraso de rastreamento no arquivo robots.txt. Se um for especificado, o rastreador usa esse atraso entre as solicitações. Caso contrário, ele usa um atraso padrão.
Identificação do agente do usuário — O rastreador usa uma string personalizada do agente do usuário para se identificar nos sites. Se necessário, os proprietários de sites podem definir regras específicas para restringir ou permitir seu rastreador.
Tratamento de erros e degradação normal — Se o arquivo robots.txt não puder ser buscado ou analisado, o rastreador prosseguirá com regras padrão conservadoras. Ele lida com erros de rede e respostas HTTP que não sejam 200.
Rastreamento limitado — Para evitar sobrecarregar o servidor, há um limite para quantas páginas podem ser rastreadas.

O script a seguir é um pseudocódigo que explica como o rastreador da web funciona:


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

Para criar um rastreador web ético e avançado que colete dados do ESG

Copie o seguinte exemplo de código para o rastreador ético avançado da Web usado neste 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)

Configure os vários atributos, incluindo agentes de usuário, coleções vazias e listas de armazenamento de dados. URLs
Ajuste as palavras-chave e os critérios de relevância no is_relevant_to_sustainable_finance() método para atender às suas necessidades específicas.
Verifique se o arquivo robots.txt permite rastrear o site e se você está usando o atraso de rastreamento e o agente de usuário especificados no arquivo robots.txt.
Considere fazer as seguintes personalizações no script de rastreador da web fornecido, conforme necessário para sua organização:
- Implemente o fetch_sitemap() método para uma descoberta de URL mais eficiente.
- Melhore o registro e o tratamento de erros para uso em produção.
- Implemente uma análise de relevância de conteúdo mais sofisticada.
- Adicione controles de profundidade e amplitude para limitar o escopo do rastreamento.

Atenção O Javascript está desativado ou não está disponível no seu navegador.

Para usar a documentação da AWS, o Javascript deve estar ativado. Consulte as páginas de Ajuda do navegador para obter instruções.

Convenções do documento

Preparando um conjunto de dados

Construindo a AWS infraestrutura