用Python抓取动态网站：全面指南

import requests
import json

def scrape_api_data(api_url, headers=None, params=None):
    try:
        response = requests.get(api_url, headers=headers, params=params)
        response.raise_for_status() # 对 HTTP 错误引发异常
        return response.json() # 假设 API 返回 JSON
    except requests.exceptions.RequestException as e:
        print(f"获取 API 数据时出错：{e}")
        return None

# 示例用法（假设产品列表的 API）
# 用网络标签中找到的实际 API URL 和参数替换
api_endpoint = "https://api.example.com/products"
custom_headers = {

{
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
    "Accept": "application/json"
}
query_params = {
    "category": "电子产品",
    "page": 1
}

data = scrape_api_data(api_endpoint, headers=custom_headers, params=query_params)
if data:
    print("成功从API抓取数据:")
    # 在此处理您的数据，例如，打印产品名称
    for item in data.get("products", [])[:3]: # 打印前3个产品
        print(f"- {item.get("name")}: ${item.get("price")}")
else:
    print("无法从API抓取数据。")

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
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
from bs4 import BeautifulSoup
import time

def scrape_with_selenium(url, wait_selector=None, scroll_to_bottom=False):
    options = Options()
    options.add_argument("--headless") # 以无头模式运行（无GUI）
    options.add_argument("--disable-gpu")
    options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36")

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

    try:
        driver.get(url)

        if wait_selector: # 等待特定元素出现
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, wait_selector))
            )
        elif scroll_to_bottom: # 处理无限滚动
            last_height = driver.execute_script("return document.body.scrollHeight")
            while True:
                driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
                time.sleep(2) # 给新内容加载时间
                new_height = driver.execute_script("return document.body.scrollHeight")
                if new_height == last_height:
                    break
                last_height = new_height

        html_content = driver.page_source
        soup = BeautifulSoup(html_content, "html.parser")
        return soup
    except Exception as e:
        print(f"Selenium抓取时出错: {e}")
        return None
    finally:
        driver.quit()

# 示例用法：
# 对于在特定元素出现后加载内容的页面
# dynamic_soup = scrape_with_selenium("https://www.example.com/dynamic-page", wait_selector=".product-list")
# if dynamic_soup:
#     print(dynamic_soup.find("h1").text)

# 对于具有无限滚动的页面
# infinite_scroll_soup = scrape_with_selenium("https://www.example.com/infinite-scroll", scroll_to_bottom=True)
# if infinite_scroll_soup:
#     print(infinite_scroll_soup.find_all("div", class_="item")[:5])

print("Selenium示例：取消注释并替换URL以进行实际使用。")

from playwright.sync_api import sync_playwright
from bs4 import BeautifulSoup
import time

def scrape_with_playwright(url, wait_selector=None, scroll_to_bottom=False):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)  # 使用 p.firefox 或 p.webkit 进行其他浏览器
        page = browser.new_page()

        try:
            page.goto(url)

            if wait_selector:  # 等待特定元素出现
                page.wait_for_selector(wait_selector, state="visible", timeout=10000)
            elif scroll_to_bottom:  # 处理无限滚动
                last_height = page.evaluate("document.body.scrollHeight")
                while True:
                    page.evaluate("window.scrollTo(0, document.body.scrollHeight);")
                    time.sleep(2)  # 给新内容加载时间
                    new_height = page.evaluate("document.body.scrollHeight")
                    if new_height == last_height:
                        break
                    last_height = new_height

            html_content = page.content()
            soup = BeautifulSoup(html_content, "html.parser")
            return soup
        except Exception as e:
            print(f"在使用 Playwright 抓取时发生错误：{e}")
            return None
        finally:
            browser.close()

# 示例用法：
# 对于在特定元素出现后加载内容的页面
# dynamic_soup_pw = scrape_with_playwright("https://www.example.com/dynamic-page", wait_selector=".data-container")
# if dynamic_soup_pw:
#     print(dynamic_soup_pw.find("h2").text)

print("Playwright 示例：取消注释并替换 URL 以进行实际使用。")

from requests_html import HTMLSession

def scrape_with_requests_html(url, sleep_time=1):
    session = HTMLSession()
    try:
        response = session.get(url)
        response.html.render(sleep=sleep_time, scrolldown=0) 
        
        return response.html
    except Exception as e:
        print(f"在 requests-html 抓取过程中发生错误: {e}")
        return None
    finally:
        session.close()

# 示例用法:
# html_obj = scrape_with_requests_html("https://www.example.com/dynamic-content-page")
# if html_obj:
#     print(html_obj.find("h1", first=True).text)

print("requests-html 示例: 取消注释并替换 URL 进行实际使用。")

## 解决方案 5: 使用 Splash 进行 JavaScript 渲染

Splash 是一个轻量级的、可脚本化的浏览器渲染服务，带有 HTTP API。它特别适用于网络爬虫，因为它可以渲染 JavaScript、处理重定向，并执行自定义 JavaScript 代码，所有这些都通过简单的 HTTP 接口实现。您可以将 Splash 作为 Docker 容器运行，方便集成到您的爬虫基础设施中。它是使用 Python 抓取动态网站的绝佳选择，当您需要可以远程控制或独立于主爬虫扩展的专用渲染服务时。

**步骤:**
1.  运行 Splash（例如，通过 Docker: `docker run -p 8050:8050 scrapinghub/splash`）。
2.  向 Splash API 发送 HTTP 请求，带有目标 URL 和渲染选项。
3.  解析返回的 HTML。

**代码示例:**

```python
import requests
from bs4 import BeautifulSoup

def scrape_with_splash(url, splash_url="http://localhost:8050/render.html"):
    try:
        params = {
            "url": url,
            "wait": 2, 
            "html": 1, 
            "timeout": 60
        }
        response = requests.get(splash_url, params=params)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")
        return soup
    except requests.exceptions.RequestException as e:
        print(f"在 Splash 抓取过程中发生错误: {e}")
        return None

# 示例用法:
# splash_soup = scrape_with_splash("https://www.example.com/dynamic-page-splash")
# if splash_soup:
#     print(splash_soup.find("title").text)

print("Splash 示例: 在使用之前确保 Splash 正在运行（例如，通过 Docker）。")

import asyncio
from pyppeteer import launch
from bs4 import BeautifulSoup

async def scrape_with_pyppeteer(url, wait_selector=None, scroll_to_bottom=False):
    browser = None

try:
        browser = await launch(headless=True)
        page = await browser.newPage()
        await page.goto(url)

        if wait_selector: # 等待特定元素出现
            await page.waitForSelector(wait_selector, {'visible': True, 'timeout': 10000})
        elif scroll_to_bottom: # 处理无限滚动
            last_height = await page.evaluate("document.body.scrollHeight")
            while True:
                await page.evaluate("window.scrollTo(0, document.body.scrollHeight);")
                await asyncio.sleep(2) # 给新内容加载时间
                new_height = await page.evaluate("document.body.scrollHeight")
                if new_height == last_height:
                    break
                last_height = new_height

        html_content = await page.content()
        soup = BeautifulSoup(html_content, "html.parser")
        return soup
    except Exception as e:
        print(f"在 Pyppeteer 抓取期间发生错误: {e}")
        return None
    finally:
        if browser:
            await browser.close()

# 示例用法（需要在异步上下文中运行）:
# async def main():
#     pyppeteer_soup = await scrape_with_pyppeteer("https://www.example.com/dynamic-pyppeteer", wait_selector=".content-area")
#     if pyppeteer_soup:
#         print(pyppeteer_soup.find("p").text)
# asyncio.run(main())

print("Pyppeteer 示例：需要在异步上下文中运行。取消注释并替换网址以进行实际使用。")

# 无限滚动的核心逻辑涉及：
# last_height = driver.execute_script("return document.body.scrollHeight")
# while True:
#     driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
#     time.sleep(sleep_time) # 根据页面加载速度调整 sleep_time
#     new_height = driver.execute_script("return document.body.scrollHeight")
#     if new_height == last_height:
#         break
#     last_height = new_height

print("无限滚动在 Selenium 和 Playwright 示例（解决方案 2 和 3）中处理。")

# 模拟用户交互（点击、输入）

# 动态网站通常需要用户交互，例如点击按钮、填写表单或选择下拉选项，以显示或加载动态内容。像 Selenium 和 Playwright 这样的浏览器自动化工具擅长模拟这些交互。通过程序控制浏览器，您可以触发 JavaScript 事件以加载所需的数据，从而使其可供抓取。这对于使用 Python 抓取动态网站至关重要，其中内容被用户操作所保护。

# 代码示例（Selenium 的点击和输入）：

from bs4 import BeautifulSoup
import time

def interact_and_scrape(url, click_selector=None, input_selector=None, input_text=None, wait_selector=None):
    options = Options()
    options.add_argument("--headless")
    options.add_argument("--disable-gpu")
    options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36")

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

    try:
        driver.get(url)

        if click_selector: # 模拟点击
            button = WebDriverWait(driver, 10).until(
                EC.element_to_be_clickable((By.CSS_SELECTOR, click_selector))
            )
            button.click()
            time.sleep(2) # 在点击后等待内容加载

        if input_selector and input_text: # 模拟文本输入
            input_field = WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, input_selector))
            )
            input_field.send_keys(input_text)
            input_field.send_keys(webdriver.Keys.RETURN) # 输入后按回车
            time.sleep(2) # 在输入后等待内容加载

        if wait_selector: # 等待新内容出现
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.CSS_SELECTOR, wait_selector))
            )

        html_content = driver.page_source
        soup = BeautifulSoup(html_content, "html.parser")
        return soup
    except Exception as e:
        print(f"交互和抓取时出错: {e}")
        return None
    finally:
        driver.quit()

# 示例用法:
# 对于具有“加载更多”按钮的页面
# interactive_soup = interact_and_scrape("https://www.example.com/products", click_selector="#load-more-btn", wait_selector=".new-product-item")
# if interactive_soup:
#     print(interactive_soup.find_all("div", class_="product-name")[:3])

# 对于搜索表单
# search_soup = interact_and_scrape("https://www.example.com/search", input_selector="#search-box", input_text="网络抓取", wait_selector=".search-results")
# if search_soup:
#     print(search_soup.find_all("li", class_="result-item")[:3])

print("Selenium 交互示例：取消注释并替换网址以进行实际使用。")

import requests
import json

def submit_dynamic_form(post_url, form_data, headers=None):
    try:
        response = requests.post(post_url, data=form_data, headers=headers)
        response.raise_for_status()
        # 根据响应，可能是 JSON 或 HTML
        try:
            return response.json()
        except json.JSONDecodeError:
            return response.text
    except requests.exceptions.RequestException as e:
        print(f"提交表单时出错: {e}")
        return None

# 示例用法（假设的搜索表单）
# 用来自网络标签的实际 POST URL、表单数据和头部进行替换
form_action_url = "https://www.example.com/api/search-results"
search_payload = {
    "query": "动态抓取",
    "category": "工具",
    "sort_by": "相关性"
}

导入 请求
导入 json

# 假设您拥有 Scrapeless API 端点和 API 密钥
SCRAPELESS_API_URL = "https://api.scrapeless.com/v1/scrape"
SCRAPELESS_API_KEY = "您的 API 密钥"

定义 scrape_dynamic_with_scrapeless(target_url, render_js=True, wait_for_selector=None, scroll_to_bottom=False):
    头 = {
        "授权": f"Bearer {SCRAPELESS_API_KEY}",
        "内容类型": "application/json"
    }
    负载 = {
        "url": target_url,
        "options": {
            "renderJavaScript": render_js,
            "waitForSelector": wait_for_selector, # 等待特定元素
            "scrollPage": scroll_to_bottom, # 模拟无限滚动
            "用户Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" # 示例用户代理
        }
    }
    尝试:
        响应 = requests.post(SCRAPELESS_API_URL, json=负载, headers=头)
        响应.raise_for_status()
        数据 = 响应.json()
        打印(f"从 {target_url} 抓取的数据：\n{数据.get('html_content')[:500]}...") # 打印 HTML 的前 500 个字符
        返回 数据
    除了 requests.exceptions.RequestException as e:
        打印(f"使用 Scrapeless 时出错：{e}")
        返回 None

# 示例用法：
# 注意：替换为实际的 Scrapeless API URL 和密钥，以及目标网址
# 为演示目的，我们将使用占位符网址
# scrape_dynamic_with_scrapeless("https://www.example.com/dynamic-data", render_js=True, wait_for_selector=".product-grid")
# scrape_dynamic_with_scrapeless("https://www.example.com/infinite-feed", render_js=True, scroll_to_bottom=True)
打印("Scrapeless 概念示例：当 renderJavaScript 为 True 时，Scrapeless 自动处理动态内容。")