기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다. # 웹 크롤러 빌드 [아키텍처](architecture.md) 섹션에 설명된 대로 애플리케이션은 각 회사에 대해 하나씩 배치로 실행됩니다. **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()` 메서드를 구현합니다. + 프로덕션 사용에 대한 오류 로깅 및 처리를 개선합니다. + 보다 정교한 콘텐츠 관련성 분석을 구현합니다. + 깊이 및 너비 제어를 추가하여 크롤링 범위를 제한합니다.