Извлечение информации с сайтов: практические решения на Python

Дополнительные примеры и продвинутые техники

В этом разделе собраны нестандартные сценарии, которые часто возникают на практике.

1. Ротация прокси и User-Agent

   Для обхода блокировок нужно подменять IP и заголовки. Пример с библиотекой fake-useragent и списком прокси.

   Пример

   import requests
   from fake_useragent import UserAgent
   from itertools import cycle
   
   proxies = [
       "http://proxy1:8080",
       "http://proxy2:8080",
   ]
   proxy_pool = cycle(proxies)
   ua = UserAgent()
   
   for url in urls:
       proxy = next(proxy_pool)
       headers = {"User-Agent": ua.random}
       try:
           response = requests.get(url, proxies={"http": proxy, "https": proxy}, headers=headers, timeout=5)
           # обработка
       except requests.exceptions.ProxyError:
           # переключение на другой прокси
           continue

   Результат: успешное получение данных, даже если один из прокси перестал работать.

2. Парсинг с помощью XPath и lxml

   XPath часто мощнее CSS-селекторов. lxml работает быстрее BeautifulSoup.

   Пример

   import requests
   from lxml import html
   
   url = "http://quotes.toscrape.com/"
   response = requests.get(url)
   tree = html.fromstring(response.content)
   quotes = tree.xpath("//div[@class='quote']/span[@class='text']/text()")
   print(quotes)

   Результат: список текстов цитат, извлечённых напрямую.

3. Асинхронная обработка с сохранением в базу данных

   Пример связки aiohttp + asyncpg для записи в PostgreSQL.

   Пример

   import asyncio
   import aiohttp
   import asyncpg
   
   async def save_to_db(pool, data):
       async with pool.acquire() as conn:
           await conn.execute("INSERT INTO quotes(text, author) VALUES($1, $2)", data["text"], data["author"])
   
   async def fetch_and_save(url):
       async with aiohttp.ClientSession() as session:
           async with session.get(url) as resp:
               html = await resp.text()
               # парсинг...
               # for each quote:
               #     await save_to_db(pool, quote_dict)
   
   async def main():
       pool = await asyncpg.create_pool("postgresql://user:pass@localhost/db")
       urls = [...]
       await asyncio.gather(*(fetch_and_save(url) for url in urls))
       await pool.close()

   Результат: данные автоматически сохраняются в таблицу, избегая дублирования через PRIMARY KEY.

4. Сбор данных через Playwright с ожиданием динамических элементов

   На примере сайта с бесконечной прокруткой.

   Пример

   from playwright.sync_api import sync_playwright
   
   with sync_playwright() as p:
       browser = p.chromium.launch()
       page = browser.new_page()
       page.goto("https://example.com/infinite-scroll")
       for _ in range(5):
           page.wait_for_timeout(2000)
           page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
       items = page.query_selector_all(".item")
       for item in items:
           print(item.inner_text())
       browser.close()

   Результат: все элементы, подгруженные через AJAX, будут доступны.

5. Обработка капчи через сервис (антигейт)

   Пример интеграции с 2captcha для разгадывания reCAPTCHA.

   Пример

   import requests
   from twocaptcha import TwoCaptcha
   
   solver = TwoCaptcha('YOUR_API_KEY')
   # после получения страницы с капчей
   result = solver.recaptcha(
       sitekey='SITE_KEY',
       url='PAGE_URL'
   )
   # подставляем result['code'] в форму и отправляем

   Результат: обход защиты, если другие способы невозможны.

Эти примеры демонстрируют, как решать реальные задачи, выходящие за рамки базовых учебников. Важно тестировать каждый блок кода на своей целевой странице и адаптировать под её структуру.