robots.txt 파일 캡처 및 처리 사이트맵 캡처 및 처리 크롤러 설계

웹 크롤러 빌드

아키텍처 섹션에 설명된 대로 애플리케이션은 각 회사에 대해 하나씩 배치로 실행됩니다.

주제

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 인프라 구축