Сбор лидов с площадок объявлений: парсер на BeautifulSoup и Selenium

import requests
from bs4 import BeautifulSoup
import csv
from datetime import datetime

def parse_job_board(url, page_count=5):
    """Сбор вакансий с доски объявлений."""
    leads = []

    for page in range(1, page_count + 1):
        params = {'page': page} if page > 1 else {}
        response = requests.get(url, params=params, headers={
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
        })
        soup = BeautifulSoup(response.text, 'html.parser')

        for item in soup.select('.job-item'):
            lead = {
                'title': item.select_one('.job-title').text.strip(),
                'company': item.select_one('.company-name').text.strip(),
                'location': item.select_one('.location').text.strip(),
                'salary': item.select_one('.salary').text.strip() if item.select_one('.salary') else '',
                'url': item.select_one('a')['href'],
                'date': datetime.now().strftime('%Y-%m-%d %H:%M'),
            }
            leads.append(lead)

    return leads

# Пример использования
# jobs = parse_job_board('https://example-board.com/jobs?keyword=python')
# print(f"Найдено вакансий: {len(jobs)}")

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
import time

def parse_dynamic_board(search_query, max_pages=3):
    """Парсинг площадки с динамической загрузкой."""
    chrome_options = Options()
    chrome_options.add_argument('--headless')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument('--disable-dev-shm-usage')

    driver = webdriver.Chrome(options=chrome_options)
    leads = []

    try:
        driver.get(f'https://example-board.com/search?q={search_query}')
        wait = WebDriverWait(driver, 10)

        # Ожидание загрузки результатов
        wait.until(EC.presence_of_all_elements_located(
            (By.CSS_SELECTOR, '.listing-item')
        ))

        # Прокрутка для загрузки всех элементов
        for _ in range(3):
            driver.execute_script('window.scrollBy(0, 500)')
            time.sleep(1)

        items = driver.find_elements(By.CSS_SELECTOR, '.listing-item')

        for item in items:
            try:
                title = item.find_element(By.CSS_SELECTOR, '.title').text
                contact = item.find_element(By.CSS_SELECTOR, '.contact').text
                leads.append({
                    'title': title,
                    'contact': contact,
                    'date': datetime.now().strftime('%Y-%m-%d %H:%M'),
                })
            except Exception:
                continue

    finally:
        driver.quit()

    return leads

import sqlite3
import re

def clean_phone(phone_text):
    """Нормализация номера телефона."""
    digits = re.findall(r'\d', phone_text)
    if len(digits) == 11 and digits[0] == '7':
        return f"+{'' .join(digits)}"
    return phone_text

def clean_email(email_text):
    """Извлечение email из текста."""
    match = re.search(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', email_text)
    return match.group(0) if match else ''

def save_unique_leads(leads):
    """Сохранение лидов с дедупликацией."""
    conn = sqlite3.connect('leads.db')
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS leads (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT,
            company TEXT,
            phone TEXT,
            email TEXT,
            source TEXT,
            url TEXT UNIQUE,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')

    inserted = 0
    for lead in leads:
        try:
            cursor.execute(
                'INSERT OR IGNORE INTO leads (title, company, phone, email, source, url) VALUES (?, ?, ?, ?, ?, ?)',
                (
                    lead.get('title', ''),
                    lead.get('company', ''),
                    clean_phone(lead.get('contact', '')),
                    clean_email(lead.get('contact', '')),
                    'job_board',
                    lead.get('url', ''),
                )
            )
            if cursor.rowcount > 0:
                inserted += 1
        except Exception:
            pass

    conn.commit()
    conn.close()
    return inserted

import csv

def export_leads_to_csv(filename='leads_export.csv'):
    """Экспорт новых лидов за сегодня в CSV."""
    conn = sqlite3.connect('leads.db')
    cursor = conn.cursor()
    cursor.execute('''
        SELECT title, company, phone, email, source, url
        FROM leads
        WHERE date(created_at) = date('now')
    ''')
    rows = cursor.fetchall()
    conn.close()

    with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.writer(f)
        writer.writerow(['Заголовок', 'Компания', 'Телефон', 'Email', 'Источник', 'URL'])
        writer.writerows(rows)

    print(f"Экспортировано {len(rows)} лидов в {filename}")
    return len(rows)