Python 网络爬虫与自动化：从 requests 到 Playwright

# 尊重 robots.txt
from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url("https://example.com/robots.txt")
rp.read()
can = rp.can_fetch("my-bot/1.0", "https://example.com/private/")
print("允许抓取?", can)

import requests

session = requests.Session()          # 复用连接 + 保持 cookie
session.headers.update({"User-Agent": "my-scraper/1.0"})

resp = session.get("https://example.com/list?page=1", timeout=10)
print(resp.status_code, resp.url)
html = resp.text                      # 默认按 header 解码
print(resp.encoding)                  # 可能是 ISO-8859-1，需要处理中文
resp.encoding = "utf-8"               # 修正编码

import httpx

# 同步
with httpx.Client(headers={"User-Agent": "x"}, timeout=10) as client:
    r = client.get("https://example.com")

# 异步（高并发）
import asyncio

async def main():
    async with httpx.AsyncClient(timeout=10) as client:
        tasks = [client.get(f"https://example.com/item/{i}") for i in range(50)]
        results = await asyncio.gather(*tasks)
        for r in results:
            if r.status_code == 200:
                process(r.text)

asyncio.run(main())

def crawl_pages(base_url, max_pages=20):
    session = requests.Session()
    for page in range(1, max_pages + 1):
        resp = session.get(f"{base_url}?page={page}", timeout=10)
        if resp.status_code != 200:
            break
        yield parse_page(resp.text)
        time.sleep(0.5)   # 限速，别打太狠

from bs4 import BeautifulSoup

soup = BeautifulSoup(html, "lxml")   # lxml 快；html.parser 更稳

# 提取标题
print(soup.title.text)

# 找单个元素
title = soup.find("h1")
# 按 class
items = soup.find_all("div", class_="product-item")
# 按属性
links = soup.find_all("a", href=True)
# CSS 选择器（更强大）
cards = soup.select("div.card > a.title")

for item in items:
    name = item.select_one("h2").text.strip()
    price = item.select_one(".price").text.strip()
    link = item.select_one("a")["href"]
    print(name, price, link)

from lxml import html as lxml_html

tree = lxml_html.fromstring(html)
# XPath
prices = tree.xpath("//div[@class='price']/text()")

# XPath 实战：提取所有商品链接与标题
links = tree.xpath("//div[@class='product']//a[@class='title']")
for a in links:
    text = a.text_content().strip()
    href = a.get("href")
    print(text, href)

import httpx, asyncio, json

async def crawl_api():
    async with httpx.AsyncClient() as client:
        # 分页 JSON API
        all_items = []
        for page in range(1, 5):
            r = await client.get(
                "https://example.com/api/items",
                params={"page": page, "limit": 100},
            )
            data = r.json()
            all_items.extend(data["items"])
            if page >= data["total_pages"]:
                break

        # 并发抓详情
        async def fetch_detail(item_id):
            r = await client.get(f"https://example.com/api/items/{item_id}")
            return r.json()

        details = await asyncio.gather(
            *(fetch_detail(i["id"]) for i in all_items[:20])
        )
        return details

data = asyncio.run(crawl_api())
print(json.dumps(data, ensure_ascii=False, indent=2)[:500])

# items.py
import scrapy

class ProductItem(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    url = scrapy.Field()

# spiders/products.py
import scrapy

class ProductSpider(scrapy.Spider):
    name = "products"
    start_urls = ["https://example.com/products"]

    def parse(self, response):
        for card in response.css("div.product"):
            item = ProductItem()
            item["name"] = card.css("h2::text").get()
            item["price"] = card.css(".price::text").get()
            item["url"] = card.css("a::attr(href)").get()
            yield item

        # 自动翻页
        next_page = response.css("a.next::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)

# 运行
scrapy crawl products -O products.json

pip install playwright
playwright install chromium

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()

    page.goto("https://example.com/dashboard", wait_until="networkidle")
    page.wait_for_selector(".data-table")      # 等元素渲染

    # 交互：点击、输入、滚动
    page.click("#load-more")
    page.fill("#search", "python")
    page.wait_for_timeout(1000)

    # 提取
    rows = page.query_selector_all("tr")
    for row in rows[:5]:
        print(row.inner_text())

    # 截图 / PDF（便于调试）
    page.screenshot(path="debug.png")
    browser.close()

from playwright.async_api import async_playwright
import asyncio

async def snap(url):
    async with async_playwright() as p:
        b = await p.chromium.launch()
        pg = await b.new_page()
        await pg.goto(url)
        title = await pg.title()
        await b.close()
        return title

async def main():
    titles = await asyncio.gather(*[snap(u) for u in URLS])
    print(titles)

# 指数退避
def fetch_with_retry(url, max_retries=3):
    for i in range(max_retries):
        try:
            return requests.get(url, timeout=10)
        except requests.RequestException as e:
            wait = 2 ** i
            print(f"重试 {i+1}，等待 {wait}s: {e}")
            time.sleep(wait)
    raise RuntimeError(f"最终失败: {url}")

import time, random

def polite_fetch(url, min_delay=1.0, max_delay=3.0):
    time.sleep(random.uniform(min_delay, max_delay))  # 随机延时更像人
    return requests.get(url, timeout=10)

proxies = [
    "http://proxy1:8080",
    "http://proxy2:8080",
]
def get_proxy():
    return {"http": random.choice(proxies), "https": random.choice(proxies)}

def clean_price(raw: str) -> float:
    """'¥1,299.00' → 1299.00"""
    import re
    s = re.sub(r"[^\d.]", "", raw)
    return float(s) if s else 0.0

def clean_text(raw: str) -> str:
    return " ".join(raw.split())   # 合并多余空白

# 结构化
records = [
    {"name": clean_text(name), "price": clean_price(price)}
    for name, price in raw_items
]

import json, sqlite3

# JSONL 追加
with open("items.jsonl", "a", encoding="utf-8") as f:
    f.write(json.dumps(record, ensure_ascii=False) + "\n")

# SQLite
conn = sqlite3.connect("items.db")
conn.execute("CREATE TABLE IF NOT EXISTS items (name TEXT, price REAL)")
conn.execute("INSERT INTO items VALUES (?, ?)", (record["name"], record["price"]))
conn.commit()