import os
import time
import argparse
import json
import requests
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import WebDriverException
from bs4 import BeautifulSoup
def create_output_folder():
folder_name = datetime.now().strftime("%Y_%m_%d-%H_%M_%S")
os.makedirs(folder_name, exist_ok=True)
return folder_name
def save_log(log_file, log_data):
with open(log_file, 'w') as f:
json.dump(log_data, f, indent=4)
def load_log(log_file):
if os.path.exists(log_file):
with open(log_file, 'r') as f:
return json.load(f)
return {"processed_domains": {}, "queue": []}
def load_proxies(file_path):
with open(file_path, 'r') as f:
return [line.strip() for line in f if line.strip()]
def check_proxy(proxy, timeout=30):
try:
if '@' in proxy:
proxy_parts = proxy.split('@')
auth, ip_port = proxy_parts[0], proxy_parts[1]
username, password = auth.split(':')
proxy_dict = {
'http': f'http://{username}:{password}@{ip_port}',
'https': f'https://{username}:{password}@{ip_port}'
}
else:
proxy_dict = {
'http': f'http://{proxy}',
'https': f'https://{proxy}'
}
response = requests.get('https://www.google.com', proxies=proxy_dict, timeout=timeout)
return response.status_code == 200
except:
return False
def get_working_proxy(proxies):
for proxy in proxies:
print(f"Проверка прокси {proxy}...")
if check_proxy(proxy):
print(f"Прокси {proxy} работает.")
return proxy
else:
print(f"Прокси {proxy} не работает. Переход к следующему.")
print("Ни один прокси не работает.")
return None
def process_domain(domain, chrome_options, parse_options):
for attempt in range(3): # Пробуем до 3 раз
try:
driver = webdriver.Chrome(options=chrome_options)
driver.set_page_load_timeout(30)
driver.get(f'https://www.similarweb.com/website/{domain}/')
time.sleep(10)
# Проверка на "Challenge validation" в заголовке страницы
title = driver.title
if "Challenge validation" in title:
print(f"Найден 'Challenge validation' на странице {domain}. Ожидание 5 секунд...")
time.sleep(5) # Ждем 5 секунд
continue # Повторяем попытку
soup = BeautifulSoup(driver.page_source, 'html.parser')
# Парсинг списка конкурентов
competitors_div = soup.find('div', {'class': 'wa-competitors__list'})
competitors = []
if competitors_div is not None:
for span in competitors_div.findAll('span', {'class': 'wa-competitors__list-item-title'}):
competitors.append(span.text.strip())
result = {"status": "success", "competitors": competitors, "attempts": attempt + 1}
# Парсинг глобального ранга
if parse_options.get('global_rank'):
global_rank_div = soup.find('div', {'class': 'wa-rank-list__item wa-rank-list__item--global'})
if global_rank_div is not None:
global_rank_value = global_rank_div.find('p', {'class': 'wa-rank-list__value'})
if global_rank_value:
result['global_rank'] = global_rank_value.text.strip()
# Парсинг Country Rank
if parse_options.get('country_rank'):
country_rank_div = soup.find('div', {'class': 'wa-rank-list__item wa-rank-list__item--country'})
if country_rank_div is not None:
country_rank_value = country_rank_div.find('p', {'class': 'wa-rank-list__value'})
country_info = country_rank_div.find('div', {'class': 'wa-rank-list__info'})
if country_rank_value and country_info:
result['country_rank'] = f"{country_rank_value.text.strip()} {country_info.text.strip()}"
# Парсинг Category Rank
if parse_options.get('category_rank'):
category_rank_div = soup.find('div', {'class': 'wa-rank-list__item wa-rank-list__item--category'})
if category_rank_div is not None:
category_rank_value = category_rank_div.find('p', {'class': 'wa-rank-list__value'})
category_info = category_rank_div.find('div', {'class': 'wa-rank-list__info'})
if category_rank_value and category_info:
result['category_rank'] = f"{category_rank_value.text.strip()} {category_info.text.strip()}"
# Парсинг Total Visits
if parse_options.get('total_visits'):
total_visits_div = soup.find('div', {'class': 'engagement-list__item'})
if total_visits_div is not None:
total_visits_value = total_visits_div.find('p', {'class': 'engagement-list__item-value'})
if total_visits_value:
result['total_visits'] = total_visits_value.text.strip()
print(f"Данные домена {domain} успешно обработаны.")
return result
except WebDriverException as e:
print(f"Ошибка при обработке домена {domain} (попытка {attempt + 1}): {e}")
time.sleep(5) # Ждем 5 секунд перед повторной попыткой
finally:
if 'driver' in locals():
driver.quit()
# Если после 3 попыток не удалось обработать домен
return {"status": "error", "error_message": str(e), "attempts": 3}
def process_domains(domains, chrome_options, output_folder, max_depth, log_file, parse_options):
log_data = load_log(log_file) # Загружаем предыдущий лог
# Если очередь пуста, инициализируем ее с переданными доменами
if not log_data["queue"]:
log_data["queue"] = [{"domain": domain, "depth": 0} for domain in domains if domain not in log_data["processed_domains"]]
while log_data["queue"]:
current_task = log_data["queue"].pop(0)
domain, depth = current_task["domain"], current_task["depth"]
if domain not in log_data["processed_domains"] and depth < max_depth:
result = process_domain(domain, chrome_options, parse_options)
log_data["processed_domains"][domain] = result # Записываем результат в лог
if result["status"] == "success":
rank_info = f"{domain}"
if 'global_rank' in result:
rank_info += f" - Global Rank: {result['global_rank']}"
if 'country_rank' in result:
rank_info += f" - Country Rank: {result['country_rank']}"
if 'category_rank' in result:
rank_info += f" - Category Rank: {result['category_rank']}"
if 'total_visits' in result:
rank_info += f" - Total Visits: {result['total_visits']}"
write_to_file(output_folder, "results.txt", rank_info)
for competitor in result["competitors"]:
if competitor not in log_data["processed_domains"]:
log_data["queue"].append({"domain": competitor, "depth": depth + 1})
save_log(log_file, log_data) # Сохраняем лог на каждом шаге
def write_to_file(folder, filename, data):
with open(os.path.join(folder, filename), 'a') as f:
f.write(data + "\n")
def read_domains_from_file(filename):
with open(filename, 'r') as f:
domains = f.read().splitlines()
return domains
def main():
parser = argparse.ArgumentParser(description='Парсинг конкурентов с SimilarWeb.')
parser.add_argument('-d', '--domain', type=str, help='Домен для парсинга.')
parser.add_argument('-l', '--list', type=str, help='Файл со списком доменов для парсинга.')
parser.add_argument('-m', '--max-depth', type=int, default=5, help='Максимальная глубина парсинга дерева конкурентов.')
parser.add_argument('-s', '--session', type=str, help='Файл с логом для продолжения сессии.')
parser.add_argument('--global-rank', action='store_true', help='Парсить Global Rank')
parser.add_argument('--country-rank', action='store_true', help='Парсить Country Rank')
parser.add_argument('--category-rank', action='store_true', help='Парсить Category Rank')
parser.add_argument('--total-visits', action='store_true', help='Парсить Total Visits')
# Добавляем новые аргументы для прокси
parser.add_argument('--proxy', type=str, help='Прокси без авторизации (ip:port)')
parser.add_argument('--proxy-cred', type=str, help='Прокси с авторизацией (ip:port@login:pass)')
parser.add_argument('--proxy-list', type=str, help='URL со списком прокси без авторизации')
parser.add_argument('--proxy-local', type=str, help='Локальный файл со списком прокси без авторизации')
parser.add_argument('--proxy-list-cred', type=str, help='URL со списком прокси с авторизацией')
parser.add_argument('--proxy-local-cred', type=str, help='Локальный файл со списком прокси с авторизацией')
args = parser.parse_args()
chrome_options = Options()
# chrome_options.headless = False # Фоновый режим
chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
# Обработка прокси
proxies = []
if args.proxy:
proxies = [args.proxy]
elif args.proxy_cred:
proxies = [args.proxy_cred]
elif args.proxy_list:
response = requests.get(args.proxy_list)
proxies = response.text.split('\n')
elif args.proxy_local:
proxies = load_proxies(args.proxy_local)
elif args.proxy_list_cred:
response = requests.get(args.proxy_list_cred)
proxies = response.text.split('\n')
elif args.proxy_local_cred:
proxies = load_proxies(args.proxy_local_cred)
working_proxy = get_working_proxy(proxies)
if working_proxy:
if '@' in working_proxy:
chrome_options.add_argument(f'--proxy-server={working_proxy.split("@")[1]}')
else:
chrome_options.add_argument(f'--proxy-server={working_proxy}')
output_folder = create_output_folder()
log_file = os.path.join(output_folder, "log.json")
if args.session:
log_file = args.session
parse_options = {
'global_rank': args.global_rank,
'country_rank': args.country_rank,
'category_rank': args.category_rank,
'total_visits': args.total_visits
}
if args.domain:
process_domains([args.domain], chrome_options, output_folder, args.max_depth, log_file, parse_options)
elif args.list:
domains = read_domains_from_file(args.list)
process_domains(domains, chrome_options, output_folder, args.max_depth, log_file, parse_options)
else:
print("Необходимо указать домен (-d) или файл со списком доменов (-l).")
print("Все данные были успешно сохранены.")
if __name__ == "__main__":
main()
