Мастерство AI-скрапинга: интеграция BrightData и Google Gemini

Создание мощного веб-скраппера с BrightData и Google Gemini

В этом руководстве показано, как создать продвинутый инструмент для веб-скрапинга, объединяющий обширную прокси-сеть BrightData с API Google Gemini для интеллектуального извлечения данных. Проект реализован на Python с использованием необходимых библиотек и аккуратного, переиспользуемого класса BrightDataScraper.

Настройка окружения

Установите все необходимые библиотеки одной командой:

!pip install langchain-brightdata langchain-google-genai langgraph langchain-core google-generativeai

Импортируйте нужные модули для работы с системой, сериализации данных и интеграции BrightData и Google Gemini:

import os
import json
from typing import Dict, Any, Optional
from langchain_brightdata import BrightDataWebScraperAPI
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.prebuilt import create_react_agent

Класс BrightDataScraper

Этот класс инкапсулирует всю логику веб-скрапинга и опционального AI.

class BrightDataScraper:
    """Расширенный веб-скраппер с использованием BrightData API"""
   
    def __init__(self, api_key: str, google_api_key: Optional[str] = None):
        """Инициализация с API-ключами"""
        self.api_key = api_key
        self.scraper = BrightDataWebScraperAPI(bright_data_api_key=api_key)
       
        if google_api_key:
            self.llm = ChatGoogleGenerativeAI(
                model="gemini-2.0-flash",
                google_api_key=google_api_key
            )
            self.agent = create_react_agent(self.llm, [self.scraper])
   
    def scrape_amazon_product(self, url: str, zipcode: str = "10001") -> Dict[str, Any]:
        """Скрапинг данных продукта Amazon"""
        try:
            results = self.scraper.invoke({
                "url": url,
                "dataset_type": "amazon_product",
                "zipcode": zipcode
            })
            return {"success": True, "data": results}
        except Exception as e:
            return {"success": False, "error": str(e)}
   
    def scrape_amazon_bestsellers(self, region: str = "in") -> Dict[str, Any]:
        """Скрапинг бестселлеров Amazon"""
        try:
            url = f"https://www.amazon.{region}/gp/bestsellers/"
            results = self.scraper.invoke({
                "url": url,
                "dataset_type": "amazon_product"
            })
            return {"success": True, "data": results}
        except Exception as e:
            return {"success": False, "error": str(e)}
   
    def scrape_linkedin_profile(self, url: str) -> Dict[str, Any]:
        """Скрапинг профиля LinkedIn"""
        try:
            results = self.scraper.invoke({
                "url": url,
                "dataset_type": "linkedin_person_profile"
            })
            return {"success": True, "data": results}
        except Exception as e:
            return {"success": False, "error": str(e)}
   
    def run_agent_query(self, query: str) -> None:
        """Запуск AI-агента с запросом на естественном языке"""
        if not hasattr(self, 'agent'):
            print("Ошибка: для работы агента требуется ключ Google API")
            return
       
        try:
            for step in self.agent.stream(
                {"messages": query},
                stream_mode="values"
            ):
                step["messages"][-1].pretty_print()
        except Exception as e:
            print(f"Ошибка агента: {e}")
   
    def print_results(self, results: Dict[str, Any], title: str = "Результаты") -> None:
        """Красивый вывод результатов"""
        print(f"\n{'='*50}")
        print(f"{title}")
        print(f"{'='*50}")
       
        if results["success"]:
            print(json.dumps(results["data"], indent=2, ensure_ascii=False))
        else:
            print(f"Ошибка: {results['error']}")
        print()

Класс позволяет удобно получать данные продуктов Amazon, списки бестселлеров, профили LinkedIn и выполнять запросы с использованием модели Gemini, с обработкой ошибок и форматированием JSON.

Основной поток выполнения

Функция main демонстрирует использование всех возможностей:

def main():
    """Основная функция выполнения"""
    BRIGHT_DATA_API_KEY = "Используйте свой API ключ"
    GOOGLE_API_KEY = "Используйте свой API ключ"
   
    scraper = BrightDataScraper(BRIGHT_DATA_API_KEY, GOOGLE_API_KEY)
   
    print(" Скрапинг бестселлеров Amazon India...")
    bestsellers = scraper.scrape_amazon_bestsellers("in")
    scraper.print_results(bestsellers, "Бестселлеры Amazon India")
   
    print(" Скрапинг продукта Amazon...")
    product_url = "https://www.amazon.com/dp/B08L5TNJHG"
    product_data = scraper.scrape_amazon_product(product_url, "10001")
    scraper.print_results(product_data, "Данные продукта Amazon")
   
    print(" Скрапинг профиля LinkedIn...")
    linkedin_url = "https://www.linkedin.com/in/satyanadella/"
    linkedin_data = scraper.scrape_linkedin_profile(linkedin_url)
    scraper.print_results(linkedin_data, "Данные профиля LinkedIn")
   
    print(" Запуск AI-агента с запросом...")
    agent_query = """
    Скрапинг данных продукта Amazon по ссылке https://www.amazon.com/dp/B0D2Q9397Y?th=1
    в Нью-Йорке (zipcode 10001) с кратким резюме ключевых деталей.
    """
    scraper.run_agent_query(agent_query)

Запуск скрипта

При запуске устанавливаются необходимые пакеты и задаются переменные окружения:

if __name__ == "__main__":
    print("Установка необходимых пакетов...")
    os.system("pip install -q langchain-brightdata langchain-google-genai langgraph")
   
    os.environ["BRIGHT_DATA_API_KEY"] = "Используйте свой API ключ"
   
    main()

Так обеспечивается установка зависимостей и конфигурация API ключей.

Расширение функционала

Эту основу можно дополнить поддержкой других типов данных, интегрировать дополнительные языковые модели или использовать скраппер в составе более крупного конвейера или сервиса. Благодаря AI и модульной архитектуре, сбор и анализ данных становится более эффективным и гибким.