| import requests |
| from bs4 import BeautifulSoup |
| from urllib.parse import urljoin, urlparse |
| import time |
|
|
| def is_same_domain(url1, url2): |
| return urlparse(url1).netloc == urlparse(url2).netloc |
|
|
| def fetch_url(url): |
| try: |
| headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'} |
| response = requests.get(url, headers=headers, timeout=10) |
| response.raise_for_status() |
| return response.text |
| except Exception as e: |
| print(f"Error fetching {url}: {e}") |
| return None |
|
|
| def extract_links(html, base_url): |
| soup = BeautifulSoup(html, 'html.parser') |
| links = set() |
| for a_tag in soup.find_all('a', href=True): |
| href = a_tag.get('href') |
| if not href: |
| continue |
| full_url = urljoin(base_url, href) |
| |
| full_url = full_url.split('#')[0] |
| if full_url.startswith('http'): |
| links.add(full_url) |
| return links |
|
|
| def extract_content(html): |
| soup = BeautifulSoup(html, 'html.parser') |
| |
| texts = [] |
| for tag in soup.find_all(['h1', 'h2', 'h3', 'p']): |
| text = tag.get_text(strip=True) |
| if text: |
| texts.append(text) |
| return texts |
|
|
| def contains_topic(texts, topic): |
| topic_lower = topic.lower() |
| matching_texts = [] |
| for text in texts: |
| if topic_lower in text.lower(): |
| matching_texts.append(text) |
| return matching_texts |
|
|
| def crawl(start_url, topic, max_depth): |
| visited = set() |
| |
| queue = [(start_url, 0)] |
| results = [] |
|
|
| while queue: |
| url, depth = queue.pop(0) |
| |
| if url in visited: |
| continue |
| |
| visited.add(url) |
| print(f"Crawling (Depth {depth}): {url}") |
| |
| html = fetch_url(url) |
| if not html: |
| continue |
| |
| texts = extract_content(html) |
| matching_texts = contains_topic(texts, topic) |
| |
| if matching_texts: |
| results.append({ |
| 'url': url, |
| 'matches': matching_texts |
| }) |
| |
| if depth < max_depth: |
| links = extract_links(html, url) |
| for link in links: |
| if link not in visited and is_same_domain(start_url, link): |
| queue.append((link, depth + 1)) |
| |
| |
| time.sleep(0.5) |
| |
| return results |
|
|
