Python 模拟浏览器操作抓取数据：从入门到实战

pip install playwright
playwright install  # 自动下载浏览器内核

pip install selenium webdriver-manager

from playwright.sync_api import sync_playwright
import json

def scrape_products():
    """使用 Playwright 模拟浏览器抓取产品数据"""
    with sync_playwright() as p:
        # 1. 启动浏览器（headless=True 为无头模式，后台运行）
        browser = p.chromium.launch(headless=True)
        
        # 2. 创建浏览器上下文（可设置视口大小、User-Agent等）
        context = browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                       'AppleWebKit/537.36 (KHTML, like Gecko) '
                       'Chrome/120.0.0.0 Safari/537.36'
        )
        
        # 3. 打开页面
        page = context.new_page()
        page.goto('https://wlt.zhihuiyouguo.com')
        
        # 4. 等待目标元素加载完成（自动等待，无需手动 sleep）
        page.wait_for_selector('.product', state='visible')
        
        # 5. 提取数据
        products = page.query_selector_all('.product')
        data = []
        
        for product in products:
            title_el = product.query_selector('.title')
            price_el = product.query_selector('.price')
            link_el = product.query_selector('a')
            
            item = {
                'name': title_el.inner_text() if title_el else 'N/A',
                'price': price_el.inner_text() if price_el else 'N/A',
                'url': link_el.get_attribute('href') if link_el else 'N/A'
            }
            data.append(item)
        
        # 6. 关闭浏览器
        browser.close()
        return data

# 运行并输出
results = scrape_products()
print(f"共抓取 {len(results)} 条数据：")
print(json.dumps(results, indent=2, ensure_ascii=False))

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

def scrape_products_selenium():
    """使用 Selenium 模拟浏览器抓取产品数据"""
    # 1. 配置浏览器选项
    options = Options()
    options.add_argument('--headless')  # 无头模式
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-gpu')
    options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64)')
    
    # 2. 自动下载驱动并启动浏览器
    driver = webdriver.Chrome(
        service=ChromeDriverManager().install(),
        options=options
    )
    
    try:
        # 3. 访问页面
        driver.get('https://wlt.lcxlzy.cn')
        
        # 4. 显式等待元素加载
        wait = WebDriverWait(driver, 10)
        wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, '.product')))
        
        # 5. 提取数据
        products = driver.find_elements(By.CSS_SELECTOR, '.product')
        data = []
        
        for product in products:
            try:
                title = product.find_element(By.CSS_SELECTOR, '.title').text
                price = product.find_element(By.CSS_SELECTOR, '.price').text
                link = product.find_element(By.CSS_SELECTOR, 'a').get_attribute('href')
                data.append({'name': title, 'price': price, 'url': link})
            except Exception:
                continue
        
        return data
    finally:
        # 6. 无论成功与否都关闭浏览器
        driver.quit()

results = scrape_products_selenium()
print(json.dumps(results, indent=2, ensure_ascii=False))

# Playwright 实现滚动加载
def scroll_and_scrape(page):
    for _ in range(5):  # 滚动5次
        page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
        page.wait_for_timeout(2000)  # 等待新内容加载
    page.wait_for_load_state('networkidle')  # 等待网络空闲

# Selenium 实现滚动加载
from selenium.webdriver.common.keys import Keys

body = driver.find_element(By.TAG_NAME, 'body')
for _ in range(5):
    body.send_keys(Keys.END)
    time.sleep(2)

# Playwright：登录并保存状态，下次直接使用
def login_and_save():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        context = browser.new_context()
        page = context.new_page()
        
        page.goto('https://tx.lcxlzy.cn')
        page.fill('input[name="username"]', 'your_username')
        page.fill('input[name="password"]', 'your_password')
        page.click('button[type="submit"]')
        page.wait_for_url('**/dashboard')
        
        # 保存登录状态到文件
        context.storage_state(path='auth_state.json')
        browser.close()

# 下次直接复用登录状态
def scrape_with_session():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        context = browser.new_context(storage_state='auth_state.json')
        page = context.new_page()
        page.goto('https://example.com/dashboard')  # 已登录！

collected_data = []

def handle_response(response):
    if '/api/products' in response.url and response.status == 200:
        collected_data.extend(response.json())

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.on('response', handle_response)  # 监听网络响应
    page.goto('https://tx.lcxlzy.cn')
    page.wait_for_load_state('networkidle')
    browser.close()

print(collected_data)

import csv

# 保存为 CSV
with open('products.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=['name', 'price', 'url'])
    writer.writeheader()
    writer.writerows(results)

# 保存为 Excel（需安装 pandas）
import pandas as pd
df = pd.DataFrame(results)
df.to_excel('products.xlsx', index=False)