Selenium WebDriver для парсинга данных: техники автоматизации

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager

# Инициализация драйвера
driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))

# Открытие страницы
driver.get('https://news-website.com')

# Поиск всех заголовков новостей
news_titles = driver.find_elements(By.CSS_SELECTOR, 'h2.news-title')

# Извлечение и вывод текста заголовков
for title in news_titles:
print(title.text)

# Закрытие браузера
driver.quit()

# Вариант 1: Автоматическая установка через webdriver-manager
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))

# Вариант 2: Ручное указание пути к драйверу
service = Service('/path/to/chromedriver')
driver = webdriver.Chrome(service=service)

# Вариант 3: Использование опций браузера
from selenium.webdriver.chrome.options import Options

chrome_options = Options()
chrome_options.add_argument("--headless") # Запуск без графического интерфейса
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("--window-size=1920x1080")

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), 
options=chrome_options)

# Получение текста элемента
element = driver.find_element(By.CSS_SELECTOR, "h1.title")
title_text = element.text # Извлекаем видимый текст

# Получение значения атрибута
link = driver.find_element(By.CSS_SELECTOR, "a.product-link")
href_value = link.get_attribute("href") # Получаем URL из атрибута href

# Получение HTML-содержимого
container = driver.find_element(By.ID, "content-container")
html_content = container.get_attribute("innerHTML") # Получаем внутренний HTML

import pandas as pd

# Находим таблицу на странице
table = driver.find_element(By.CSS_SELECTOR, "table.data-table")

# Извлекаем заголовки таблицы
headers = []
header_cells = table.find_elements(By.CSS_SELECTOR, "th")
for cell in header_cells:
headers.append(cell.text)

# Извлекаем строки данных
rows_data = []
rows = table.find_elements(By.CSS_SELECTOR, "tr")[1:] # Пропускаем строку заголовка
for row in rows:
cells = row.find_elements(By.CSS_SELECTOR, "td")
row_data = [cell.text for cell in cells]
rows_data.append(row_data)

# Создаем DataFrame из собранных данных
df = pd.DataFrame(rows_data, columns=headers)

# Экспортируем в Excel
df.to_excel("extracted_data.xlsx", index=False)

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
import time

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))

# Устанавливаем неявное ожидание для всего драйвера
driver.implicitly_wait(10)

# Открываем страницу с динамическим контентом
driver.get("https://example-dynamic-site.com/products")

# Явное ожидание загрузки контейнера с продуктами
products_container = WebDriverWait(driver, 20).until(
EC.presence_of_element_located((By.CLASS_NAME, "products-grid"))
)

# Прокрутка страницы для загрузки ленивых изображений
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(2) # Даём время на загрузку

# Получаем список всех продуктов
product_cards = products_container.find_elements(By.CLASS_NAME, "product-card")

# Собираем данные из карточек продуктов
products_data = []
for card in product_cards:
try:
name = card.find_element(By.CLASS_NAME, "product-name").text
price = card.find_element(By.CLASS_NAME, "product-price").text

# Клик для раскрытия дополнительной информации
details_button = card.find_element(By.CLASS_NAME, "details-btn")
driver.execute_script("arguments[0].click();", details_button)

# Ждём появления модального окна
modal = WebDriverWait(driver, 10).until(
EC.visibility_of_element_located((By.CLASS_NAME, "product-details-modal"))
)

# Извлекаем дополнительные данные
description = modal.find_element(By.CLASS_NAME, "description").text
specifications = modal.find_element(By.CLASS_NAME, "specifications").text

# Закрываем модальное окно
close_btn = modal.find_element(By.CLASS_NAME, "close-modal")
driver.execute_script("arguments[0].click();", close_btn)

# Ждём исчезновения модального окна
WebDriverWait(driver, 10).until(
EC.invisibility_of_element_located((By.CLASS_NAME, "product-details-modal"))
)

# Сохраняем данные
products_data.append({
"name": name,
"price": price,
"description": description,
"specifications": specifications
})

except Exception as e:
print(f"Error processing product: {e}")

# Вывод собранных данных
for product in products_data:
print(f"Product: {product['name']}")
print(f"Price: {product['price']}")
print(f"Description: {product['description'][:100]}...")
print("-" * 50)

driver.quit()

# Функция для прокрутки страницы с бесконечной подгрузкой
def scroll_to_bottom(driver, max_scrolls=10):
scrolls = 0
last_height = driver.execute_script("return document.body.scrollHeight")

while scrolls < max_scrolls:
# Прокручиваем вниз
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")

# Ждем загрузки новых элементов
time.sleep(3)

# Проверяем, изменилась ли высота страницы
new_height = driver.execute_script("return document.body.scrollHeight")
if new_height == last_height:
break

last_height = new_height
scrolls += 1

# Выводим прогресс
print(f"Scroll {scrolls}/{max_scrolls} completed")

# Использование функции
driver.get("https://infinite-scroll-example.com")
scroll_to_bottom(driver, max_scrolls=5) # Прокручиваем 5 раз

# Теперь собираем все загруженные элементы
items = driver.find_elements(By.CSS_SELECTOR, ".item")
print(f"Total items loaded: {len(items)}")

import pandas as pd
import matplotlib.pyplot as plt
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager
from datetime import datetime
import time
import os
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

class DataCollector:
def __init__(self):
options = webdriver.ChromeOptions()
options.add_argument('--headless')
options.add_argument('--disable-gpu')
self.driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), 
options=options)
self.timestamp = datetime.now().strftime("%Y%m%d_%H%M")

# Создаем директорию для сохранения данных
if not os.path.exists('data'):
os.makedirs('data')
if not os.path.exists('reports'):
os.makedirs('reports')

def collect_product_data(self, urls):
"""Собирает данные о продуктах с указанных URL"""
all_products = []

for url in urls:
try:
self.driver.get(url)
time.sleep(3) # Ждем загрузку страницы

# Находим все карточки продуктов
products = self.driver.find_elements(By.CSS_SELECTOR, '.product-item')

for product in products:
try:
name = product.find_element(By.CSS_SELECTOR, '.product-name').text
price_elem = product.find_element(By.CSS_SELECTOR, '.product-price')
price = price_elem.text.replace('$', '').replace(',', '').strip()
rating_elem = product.find_elements(By.CSS_SELECTOR, '.rating-value')
rating = rating_elem[0].text if rating_elem else "N/A"

all_products.append({
'name': name,
'price': float(price) if price.replace('.', '', 1).isdigit() else 0,
'rating': float(rating) if rating != "N/A" and rating.replace('.', '', 1).isdigit() else 0,
'url': url,
'timestamp': self.timestamp
})
except Exception as e:
print(f"Error extracting product data: {e}")

except Exception as e:
print(f"Error processing URL {url}: {e}")

# Сохраняем собранные данные
df = pd.DataFrame(all_products)
csv_path = f'data/products_{self.timestamp}.csv'
df.to_csv(csv_path, index=False)
print(f"Collected {len(all_products)} products. Data saved to {csv_path}")

return df

def analyze_data(self, df):
"""Анализирует собранные данные и создает отчет"""
if df.empty:
print("No data to analyze")
return None

# Базовый анализ
analysis = {
'total_products': len(df),
'avg_price': df['price'].mean(),
'max_price': df['price'].max(),
'min_price': df['price'].min(),
'avg_rating': df['rating'].mean() if 'rating' in df.columns else 'N/A'
}

# Создаем простые визуализации
plt.figure(figsize=(12, 6))

# Гистограмма цен
plt.subplot(1, 2, 1)
df['price'].hist(bins=20)
plt.title('Price Distribution')
plt.xlabel('Price')
plt.ylabel('Count')

# Диаграмма "цена vs рейтинг"
if 'rating' in df.columns:
plt.subplot(1, 2, 2)
plt.scatter(df['price'], df['rating'])
plt.title('Price vs Rating')
plt.xlabel('Price')
plt.ylabel('Rating')

# Сохраняем визуализации
chart_path = f'reports/analysis_chart_{self.timestamp}.png'
plt.tight_layout()
plt.savefig(chart_path)

# Создаем текстовый отчет
report_text = f"""
Product Analysis Report – {datetime.now().strftime("%Y-%m-%d %H:%M")}

Summary Statistics:
- Total Products: {analysis['total_products']}
- Average Price: ${analysis['avg_price']:.2f}
- Maximum Price: ${analysis['max_price']:.2f}
- Minimum Price: ${analysis['min_price']:.2f}
- Average Rating: {analysis['avg_rating'] if analysis['avg_rating'] != 'N/A' else 'N/A'}

Top 5 Most Expensive Products:
{df.sort_values('price', ascending=False).head(5)[['name', 'price']].to_string(index=False)}

See attached charts for further analysis.
"""

report_path = f'reports/report_{self.timestamp}.txt'
with open(report_path, 'w') as f:
f.write(report_text)

print(f"Analysis completed. Report saved to {report_path}")
return {
'analysis': analysis,
'chart_path': chart_path,
'report_path': report_path,
'csv_path': f'data/products_{self.timestamp}.csv'
}

def send_email_report(self, report_data, recipient):
"""Отправляет отчет по электронной почте"""
if not report_data:
print("No report data to send")
return

try:
# Настройка email
msg = MIMEMultipart()
msg['From'] = 'your_email@example.com'
msg['To'] = recipient
msg['Subject'] = f'Product Analysis Report – {datetime.now().strftime("%Y-%m-%d")}'

# Добавляем текст отчета
with open(report_data['report_path'], 'r') as f:
report_text = f.read()
msg.attach(MIMEText(report_text, 'plain'))

# Прикрепляем график
with open(report_data['chart_path'], 'rb') as f:
attachment = MIMEApplication(f.read(), _subtype="png")
attachment.add_header('Content-Disposition', 'attachment', 
filename=os.path.basename(report_data['chart_path']))
msg.attach(attachment)

# Прикрепляем CSV с данными
with open(report_data['csv_path'], 'rb') as f:
attachment = MIMEApplication(f.read(), _subtype="csv")
attachment.add_header('Content-Disposition', 'attachment', 
filename=os.path.basename(report_data['csv_path']))
msg.attach(attachment)

# Отправка (для реального использования настройте свой SMTP-сервер)
print(f"Email report would be sent to {recipient} (simulation)")
# Раскомментируйте для реальной отправки
# with smtplib.SMTP('smtp.example.com', 587) as server:
# server.starttls()
# server.login('your_email@example.com', 'password')
# server.send_message(msg)

except Exception as e:
print(f"Error sending email: {e}")

def close(self):
"""Закрывает браузер и освобождает ресурсы"""
self.driver.quit()

# Пример использования
if __name__ == "__main__":
collector = DataCollector()
try:
# Список URL для сбора данных
urls = [
"https://example-shop.com/category/electronics",
"https://example-shop.com/category/computers"
]

# Сбор данных
data = collector.collect_product_data(urls)

# Анализ данных
report = collector.analyze_data(data)

# Отправка отчета
if report:
collector.send_email_report(report, "analyst@example.com")

finally:
collector.close()

# Откройте редактор crontab
crontab -e

# Добавьте строку для запуска скрипта каждый день в 8:00 утра
0 8 * * * cd /path/to/project && /usr/bin/python3 data_collector.py >> /path/to/logs/collector.log 2>&1