452 lines
18 KiB
Python
Executable file
452 lines
18 KiB
Python
Executable file
#!/usr/bin/env python
|
|
|
|
import sys
|
|
import re
|
|
import os
|
|
import requests
|
|
import phonenumbers
|
|
from bs4 import BeautifulSoup
|
|
from urllib.parse import urljoin, urlparse, urldefrag
|
|
import csv
|
|
from collections import deque
|
|
import spacy
|
|
import argparse
|
|
from xml.etree import ElementTree as ET
|
|
from datetime import datetime
|
|
import itertools
|
|
import gc
|
|
import time
|
|
import validators
|
|
import tldextract
|
|
|
|
# Load spaCy model for NER
|
|
nlp = spacy.load("en_core_web_sm")
|
|
|
|
# List of social media domains and their URL shorteners
|
|
SOCIAL_MEDIA_DOMAINS = [
|
|
'twitter.com', 't.co', 'x.com', 'facebook.com', 'fb.com', 'youtube.com',
|
|
'youtu.be', 'instagram.com', 'pinterest.com', 'linkedin.com', 'lnkd.in', 'tiktok.com'
|
|
]
|
|
|
|
# Regular expressions for email addresses
|
|
EMAIL_REGEX = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b')
|
|
|
|
META_INFO_FILE = 'meta-info.csv'
|
|
STATUS_FILE = '.status'
|
|
|
|
# Network request timeout and retry settings
|
|
TIMEOUT = 60 # 1 minute timeout
|
|
MAX_RETRIES = 3
|
|
|
|
def is_valid_url(url):
|
|
return validators.url(url)
|
|
|
|
def get_root_domain(url):
|
|
return tldextract.extract(url).registered_domain
|
|
|
|
def is_social_media_url(url):
|
|
root_domain = get_root_domain(url)
|
|
return any(domain in root_domain for domain in SOCIAL_MEDIA_DOMAINS)
|
|
|
|
def fetch_with_retry(url, timeout=TIMEOUT, max_retries=MAX_RETRIES):
|
|
for attempt in range(max_retries):
|
|
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=timeout, allow_redirects=True)
|
|
response.raise_for_status()
|
|
return response
|
|
except requests.RequestException as e:
|
|
print(f"Attempt {attempt + 1} failed: {e}")
|
|
if attempt == max_retries - 1:
|
|
raise
|
|
|
|
def fetch_sitemaps(base_url):
|
|
sitemaps = set()
|
|
|
|
robots_url = urljoin(base_url, '/robots.txt')
|
|
print(f"Fetching {robots_url}")
|
|
try:
|
|
response = fetch_with_retry(robots_url)
|
|
for line in response.text.split('\n'):
|
|
if line.strip().lower().startswith('sitemap:'):
|
|
sitemap_url = line.split(':', 1)[1].strip()
|
|
if is_valid_url(sitemap_url):
|
|
print(f"Found sitemap in robots.txt: {sitemap_url}")
|
|
sitemaps.add(sitemap_url)
|
|
except requests.RequestException as e:
|
|
print(f"Error fetching robots.txt: {e}")
|
|
|
|
common_sitemap_url = urljoin(base_url, '/sitemap.xml')
|
|
print(f"Checking common location for sitemap: {common_sitemap_url}")
|
|
try:
|
|
response = fetch_with_retry(common_sitemap_url)
|
|
print(f"Found sitemap at common location: {common_sitemap_url}")
|
|
sitemaps.add(common_sitemap_url)
|
|
except requests.RequestException as e:
|
|
print(f"Error fetching sitemap.xml: {e}")
|
|
|
|
return sitemaps
|
|
|
|
def parse_sitemap(sitemap_url):
|
|
print(f"Parsing sitemap: {sitemap_url}")
|
|
try:
|
|
response = fetch_with_retry(sitemap_url)
|
|
root = ET.fromstring(response.content)
|
|
for url in root.findall('.//{http://www.sitemaps.org/schemas/sitemap/0.9}loc'):
|
|
if url is not None and is_valid_url(url.text):
|
|
yield url.text
|
|
print(f"Found URL in sitemap: {url.text}")
|
|
except requests.RequestException as e:
|
|
print(f"Error fetching sitemap: {e}")
|
|
except ET.ParseError as e:
|
|
print(f"Error parsing sitemap: {e}")
|
|
|
|
def get_all_links_from_sitemaps(base_url):
|
|
sitemap_urls = fetch_sitemaps(base_url)
|
|
all_urls = set()
|
|
|
|
for sitemap_url in sitemap_urls:
|
|
all_urls.update(parse_sitemap(sitemap_url))
|
|
|
|
print(f"Total URLs found in sitemaps: {len(all_urls)}")
|
|
return all_urls
|
|
|
|
def get_seo_meta(soup, base_url):
|
|
title = soup.title.string if soup.title else ''
|
|
meta_description = ''
|
|
meta_robots = ''
|
|
meta_keywords = ''
|
|
canonical = ''
|
|
google_analytics = 'no'
|
|
google_ads_tag = 'no'
|
|
google_tag_manager = 'no'
|
|
bing_analytics = 'no'
|
|
schema_found = 'no'
|
|
first_h1 = ''
|
|
outbound_external = set()
|
|
outbound_internal = set()
|
|
word_count = 0
|
|
phone_numbers = []
|
|
email_addresses = []
|
|
|
|
for meta in soup.find_all('meta'):
|
|
if meta.get('name') == 'description':
|
|
meta_description = meta.get('content', '')
|
|
if meta.get('name') == 'robots':
|
|
meta_robots = meta.get('content', '')
|
|
if meta.get('name') == 'keywords':
|
|
meta_keywords = meta.get('content', '')
|
|
|
|
canonical_link = soup.find('link', rel='canonical')
|
|
if canonical_link:
|
|
canonical = canonical_link.get('href', '')
|
|
|
|
for script in soup.find_all('script'):
|
|
if script.string:
|
|
script_content = script.string.lower()
|
|
if 'gtag(' in script_content and 'ua-' in script_content:
|
|
google_analytics = 'yes'
|
|
if 'gtag(' in script_content and 'aw-' in script_content:
|
|
google_ads_tag = 'yes'
|
|
if 'googletagmanager.com' in script_content:
|
|
google_tag_manager = 'yes'
|
|
if 'bat.bing.com' in script_content or 'bing.com/tag.js' in script_content:
|
|
bing_analytics = 'yes'
|
|
if 'schema.org' in script_content:
|
|
schema_found = 'yes'
|
|
|
|
h1_tag = soup.find('h1')
|
|
if h1_tag:
|
|
first_h1 = h1_tag.get_text(strip=True)
|
|
|
|
body_content = soup.body.get_text(separator=' ') if soup.body else ''
|
|
word_count = len(body_content.split())
|
|
|
|
for match in phonenumbers.PhoneNumberMatcher(body_content, "US"):
|
|
phone_number = phonenumbers.format_number(match.number, phonenumbers.PhoneNumberFormat.E164)
|
|
anchor_text = ''
|
|
paragraph_text = ''
|
|
span_text = ''
|
|
|
|
anchor = soup.find('a', string=re.compile(re.escape(match.raw_string)))
|
|
if anchor:
|
|
anchor_text = anchor.get_text()
|
|
|
|
paragraph = soup.find('p', string=re.compile(re.escape(match.raw_string)))
|
|
if paragraph:
|
|
paragraph_text = paragraph.get_text()
|
|
|
|
span = soup.find('span', string=re.compile(re.escape(match.raw_string)))
|
|
if span:
|
|
span_text = span.get_text()
|
|
|
|
phone_numbers.append((phone_number, anchor_text, paragraph_text, span_text))
|
|
|
|
for email in EMAIL_REGEX.findall(body_content):
|
|
email_domain = email.split('@')[-1]
|
|
email_addresses.append((email, email_domain))
|
|
|
|
for link in soup.body.find_all('a', href=True) if soup.body else []:
|
|
full_link = urldefrag(urljoin(base_url, link['href']))[0]
|
|
if is_valid_url(full_link):
|
|
if get_root_domain(full_link) != get_root_domain(base_url):
|
|
outbound_external.add(full_link)
|
|
else:
|
|
outbound_internal.add(full_link)
|
|
|
|
return (title, meta_description, meta_robots, meta_keywords, google_analytics, google_ads_tag,
|
|
google_tag_manager, bing_analytics, schema_found, canonical, first_h1, word_count, len(outbound_external),
|
|
len(outbound_internal), phone_numbers, email_addresses)
|
|
|
|
def get_all_links(url, base_url):
|
|
print(f"Fetching {url}")
|
|
try:
|
|
response = fetch_with_retry(url)
|
|
final_url = response.url
|
|
status_code = response.status_code
|
|
print(f"Status Code: {status_code} for {url}")
|
|
|
|
if status_code == 200:
|
|
if url.endswith('.xml'):
|
|
return parse_sitemap(url)
|
|
else:
|
|
soup = BeautifulSoup(response.content, "html.parser")
|
|
links = [a.get('href') for a in soup.find_all('a', href=True)]
|
|
print(f"Found {len(links)} links on {url}")
|
|
|
|
(title, meta_description, meta_robots, meta_keywords, google_analytics, google_ads_tag,
|
|
google_tag_manager, bing_analytics, schema_found, canonical, first_h1, word_count,
|
|
outbound_external, outbound_internal, phone_numbers,
|
|
email_addresses) = get_seo_meta(soup, base_url)
|
|
|
|
internal_links = set()
|
|
external_links = []
|
|
social_media_links = []
|
|
for link in links:
|
|
full_link = urldefrag(urljoin(base_url, link))[0]
|
|
if is_valid_url(full_link):
|
|
if get_root_domain(full_link) == get_root_domain(base_url):
|
|
internal_links.add(full_link)
|
|
else:
|
|
if is_social_media_url(full_link):
|
|
social_media_links.append((full_link.lower(), get_root_domain(full_link), url))
|
|
else:
|
|
external_links.append((full_link.lower(), get_root_domain(full_link), url))
|
|
|
|
print(f"Filtered to {len(internal_links)} internal links on the same domain")
|
|
print(f"Found {len(external_links)} external links")
|
|
print(f"Found {len(social_media_links)} social media links")
|
|
|
|
return (internal_links, external_links, social_media_links, url, status_code, final_url, title,
|
|
meta_description, meta_robots, meta_keywords, google_analytics, google_ads_tag,
|
|
google_tag_manager, bing_analytics, schema_found, canonical, first_h1, word_count,
|
|
outbound_external, outbound_internal, phone_numbers, email_addresses)
|
|
else:
|
|
return set(), [], [], url, status_code, final_url, '', '', '', '', 'no', 'no', 'no', 'no', 'no', '', '', 0, 0, 0, [], []
|
|
|
|
except requests.RequestException as e:
|
|
print(f"Error fetching {url}: {e}")
|
|
return set(), [], [], url, "Error", url, '', '', '', '', 'no', 'no', 'no', 'no', 'no', '', '', 0, 0, 0, [], []
|
|
|
|
def initialize_meta_info(output_dir):
|
|
meta_info_path = os.path.join(output_dir, META_INFO_FILE)
|
|
if os.path.exists(meta_info_path):
|
|
with open(meta_info_path, mode='r', newline='') as file:
|
|
reader = csv.DictReader(file)
|
|
return {row['url']: row for row in reader}
|
|
else:
|
|
with open(meta_info_path, mode='w', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(['url', 'status', 'found_date', 'last_visited_date', 'retry_count'])
|
|
return {}
|
|
|
|
def update_meta_info(meta_info, output_dir):
|
|
meta_info_path = os.path.join(output_dir, META_INFO_FILE)
|
|
with open(meta_info_path, mode='w', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(['url', 'status', 'found_date', 'last_visited_date', 'retry_count'])
|
|
for url, info in meta_info.items():
|
|
writer.writerow([url, info['status'], info['found_date'], info['last_visited_date'], info['retry_count']])
|
|
|
|
def add_url_to_meta_info(meta_info, url, output_dir):
|
|
if url not in meta_info:
|
|
meta_info[url] = {
|
|
'url': url,
|
|
'status': 'pending',
|
|
'found_date': datetime.now().isoformat(),
|
|
'last_visited_date': '',
|
|
'retry_count': 0
|
|
}
|
|
meta_info_path = os.path.join(output_dir, META_INFO_FILE)
|
|
with open(meta_info_path, mode='a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow([url, 'pending', datetime.now().isoformat(), '', 0])
|
|
|
|
def mark_url_visited(meta_info, url, output_dir):
|
|
if url in meta_info:
|
|
meta_info[url]['status'] = 'visited'
|
|
meta_info[url]['last_visited_date'] = datetime.now().isoformat()
|
|
update_meta_info(meta_info, output_dir)
|
|
|
|
def mark_url_error(meta_info, url, output_dir):
|
|
if url in meta_info:
|
|
meta_info[url]['status'] = 'error'
|
|
meta_info[url]['retry_count'] += 1
|
|
meta_info[url]['last_visited_date'] = datetime.now().isoformat()
|
|
update_meta_info(meta_info, output_dir)
|
|
|
|
def check_status_file(output_dir):
|
|
status_file_path = os.path.join(output_dir, STATUS_FILE)
|
|
if os.path.exists(status_file_path):
|
|
with open(status_file_path, 'r') as file:
|
|
content = file.read().strip()
|
|
if content == "completed":
|
|
return True
|
|
return False
|
|
|
|
def mark_as_completed(output_dir):
|
|
status_file_path = os.path.join(output_dir, STATUS_FILE)
|
|
with open(status_file_path, 'w') as file:
|
|
file.write("completed")
|
|
|
|
def print_meta_info_summary(meta_info):
|
|
pending_count = sum(1 for info in meta_info.values() if info['status'] == 'pending')
|
|
visited_count = sum(1 for info in meta_info.values() if info['status'] == 'visited')
|
|
error_count = sum(1 for info in meta_info.values() if info['status'] == 'error')
|
|
|
|
print(f"Found {pending_count} pending pages.")
|
|
print(f"Found {visited_count} visited pages.")
|
|
print(f"Found {error_count} pages with errors.")
|
|
print("Resuming from previous state...")
|
|
|
|
def scrape_website(base_url, output_dir):
|
|
if check_status_file(output_dir):
|
|
print("Scrape already completed for this domain.")
|
|
return
|
|
|
|
visited = set()
|
|
to_visit = deque()
|
|
|
|
meta_info = initialize_meta_info(output_dir)
|
|
print_meta_info_summary(meta_info)
|
|
|
|
if not meta_info:
|
|
initial_urls = get_all_links_from_sitemaps(base_url)
|
|
for url in initial_urls:
|
|
add_url_to_meta_info(meta_info, url, output_dir)
|
|
|
|
for url, info in meta_info.items():
|
|
if info['status'] == 'pending':
|
|
to_visit.append(url)
|
|
elif info['status'] == 'visited':
|
|
visited.add(url)
|
|
|
|
internal_csv_file = os.path.join(output_dir, 'pages.csv')
|
|
external_csv_file = os.path.join(output_dir, 'external-links.csv')
|
|
social_media_csv_file = os.path.join(output_dir, 'social-media-links.csv')
|
|
phone_csv_file = os.path.join(output_dir, 'phone-numbers.csv')
|
|
email_csv_file = os.path.join(output_dir, 'email-addresses.csv')
|
|
|
|
# Initialize CSV files with headers
|
|
csv_files = [
|
|
(internal_csv_file, ['found url', 'status', 'final url', 'canonical', 'title', 'first h1',
|
|
'meta description', 'meta robots', 'meta keywords', 'google analytics',
|
|
'google ads tag', 'google tag manager', 'bing analytics', 'schema found',
|
|
'word count', 'outbound external', 'outbound internal']),
|
|
(external_csv_file, ['external link', 'linking domain', 'linking page']),
|
|
(social_media_csv_file, ['external link', 'linking domain', 'linking page']),
|
|
(phone_csv_file, ['phone number', 'anchor', 'paragraph', 'span', 'found on page']),
|
|
(email_csv_file, ['email address', 'email domain', 'found on page'])
|
|
]
|
|
|
|
for file_path, header in csv_files:
|
|
if not os.path.exists(file_path):
|
|
with open(file_path, 'w', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(header)
|
|
|
|
processed_count = 0
|
|
while to_visit:
|
|
current_url = to_visit.popleft()
|
|
if current_url not in visited:
|
|
visited.add(current_url)
|
|
print(f"Visiting: {current_url}")
|
|
|
|
if current_url.endswith('.xml'):
|
|
internal_links = parse_sitemap(current_url)
|
|
for link in internal_links:
|
|
if link not in meta_info:
|
|
add_url_to_meta_info(meta_info, link, output_dir)
|
|
to_visit.append(link)
|
|
else:
|
|
(internal_links, external_links, social_media_links, found_url, status_code, final_url, title,
|
|
meta_description, meta_robots, meta_keywords, google_analytics, google_ads_tag,
|
|
google_tag_manager, bing_analytics, schema_found, canonical, first_h1, word_count,
|
|
outbound_external, outbound_internal, phone_numbers, email_addresses) = get_all_links(current_url, base_url)
|
|
|
|
mark_url_visited(meta_info, current_url, output_dir)
|
|
|
|
new_internal_links = internal_links - visited
|
|
if new_internal_links:
|
|
print(f"Adding {len(new_internal_links)} new internal links to visit")
|
|
to_visit.extend(new_internal_links)
|
|
else:
|
|
print(f"No new internal links found on {current_url}")
|
|
|
|
# Stream write to CSV files
|
|
with open(internal_csv_file, 'a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow([found_url, status_code, final_url, canonical, title, first_h1, meta_description,
|
|
meta_robots, meta_keywords, google_analytics, google_ads_tag, google_tag_manager,
|
|
bing_analytics, schema_found, word_count, outbound_external, outbound_internal])
|
|
|
|
with open(external_csv_file, 'a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
for ext_link, domain, page in external_links:
|
|
writer.writerow([ext_link, domain, page])
|
|
|
|
with open(social_media_csv_file, 'a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
for soc_link, domain, page in social_media_links:
|
|
writer.writerow([soc_link, domain, page])
|
|
|
|
with open(phone_csv_file, 'a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
for phone, anchor, paragraph, span in phone_numbers:
|
|
writer.writerow([phone, anchor, paragraph, span, found_url])
|
|
|
|
with open(email_csv_file, 'a', newline='') as file:
|
|
writer = csv.writer(file)
|
|
for email, domain in email_addresses:
|
|
writer.writerow([email, domain, found_url])
|
|
|
|
processed_count += 1
|
|
if processed_count % 100 == 0:
|
|
print(f"Processed {processed_count} pages. Clearing memory...")
|
|
gc.collect() # Manual garbage collection
|
|
|
|
mark_as_completed(output_dir)
|
|
update_meta_info(meta_info, output_dir)
|
|
print("No more links to visit. Finished scraping.")
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description='Scrape website pages and extract information.')
|
|
parser.add_argument('base_url', help='The base URL of the website to scrape')
|
|
parser.add_argument('-o', '--output', help='The directory to save the CSV files', default='.')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.base_url or not is_valid_url(args.base_url):
|
|
print("Please provide a valid URL.")
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
base_url = args.base_url
|
|
output_dir = args.output
|
|
|
|
print(f"Starting to scrape {base_url}")
|
|
scrape_website(base_url, output_dir)
|
|
print("Finished scraping.")
|