构建一个电子商务抓取程序：一个实用的 Python 管道

©2026 NST LABS TECH LTD. 保留所有权利。

import httpx

def fetch_html(url: str) -> str:
    response = httpx.get(
        url,
        headers={"User-Agent": "CatalogQA/1.0 contact@example.com"},
        timeout=20,
        follow_redirects=True,
    )
    response.raise_for_status()
    if "text/html" not in response.headers.get("content-type", ""):
        raise ValueError("意外的内容类型")
    return response.text

import json
from bs4 import BeautifulSoup

def product_nodes(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    nodes = []
    for script in soup.select('script[type="application/ld+json"]'):
        try:
            value = json.loads(script.string or "null")
        except json.JSONDecodeError:
            continue
        values = value if isinstance(value, list) else [value]
        for item in values:
            if isinstance(item, dict) and item.get("@type") == "Product":
                nodes.append(item)
    return nodes

from playwright.sync_api import sync_playwright

def rendered_product(url: str) -> tuple[str, str, str]:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1440, "height": 900})
        page.goto(url, wait_until="domcontentloaded", timeout=30_000)
        page.locator('[data-testid="product-title"]').wait_for(timeout=10_000)
        result = (page.url, page.title(), page.content())
        browser.close()
        return result

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
  observation_key TEXT PRIMARY KEY,
  canonical_url TEXT NOT NULL,
  source_product_id TEXT NOT NULL,
  price_text TEXT,
  currency TEXT,
  availability_text TEXT,
  observed_at TEXT NOT NULL,
  content_hash TEXT NOT NULL,
  accepted INTEGER NOT NULL CHECK (accepted IN (0, 1))
)
"""

with sqlite3.connect("products.db") as db:
    db.execute(SCHEMA)