使用Python进行Shopify网页抓取：商品、变体与CSV导出

python3 -m venv .venv
source .venv/bin/activate
python -m pip install requests==2.34.2 beautifulsoup4==4.15.0

py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install requests==2.34.2 beautifulsoup4==4.15.0

.\.venv\Scripts\python.exe -m pip install requests==2.34.2 beautifulsoup4==4.15.0

.\.venv\Scripts\python.exe shopify_scraper.py

import csv
import json
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from decimal import Decimal
from pathlib import Path

import requests

BASE = "https://theme-dawn-demo.myshopify.com"
HANDLE = "puff-olive-leaf"
OUT = Path("results")


def make_session():
    session = requests.Session()
    session.trust_env = False
    session.headers["User-Agent"] = "ShopifyTutorial/1.0"
    return session


def retry_after_message(value):
    if value:
        try:
            if value.strip().isdigit():
                seconds = int(value.strip())
            else:
                deadline = parsedate_to_datetime(value)
                if deadline.tzinfo is None:
                    raise ValueError("Retry-After date must include a timezone")
                seconds = max(0, int((deadline - datetime.now(timezone.utc)).total_seconds()) + 1)
            return f"Wait at least {seconds} seconds before considering a rerun."
        except (TypeError, ValueError, OverflowError):
            pass
    return "Pause collection and confirm the permitted rate before rerunning."


def get_response(session, url, **kwargs):
    try:
        response = session.get(url, timeout=(10, 30), allow_redirects=False, **kwargs)
    except requests.exceptions.ProxyError:
        raise RuntimeError("Proxy connection failed; check gateway and authentication.") from None
    except requests.exceptions.SSLError:
        raise RuntimeError("TLS verification failed; check the certificate and trust configuration.") from None
    except requests.exceptions.Timeout:
        raise RuntimeError("Request timed out; check connectivity before rerunning.") from None
    except requests.exceptions.ConnectionError:
        raise RuntimeError("Connection failed; check DNS, host, port, and network access.") from None
    except requests.exceptions.RequestException:
        raise RuntimeError("Request failed; inspect sanitized diagnostics.") from None

    status = response.status_code
    if 300 <= status < 400:
        raise RuntimeError("Redirect received; verify the destination before changing the URL.")
    if status == 429:
        raise RuntimeError("HTTP 429: " + retry_after_message(response.headers.get("Retry-After")))
    messages = {
        403: "Access denied; stop and confirm the permitted access method.",
        407: "Proxy authentication failed; check the supported authentication mode.",
        404: "Not found; verify the handle, locale, and endpoint.",
    }
    if status >= 400:
        raise RuntimeError(f"HTTP {status}: " + messages.get(status, "Stop and review the response."))
    return response


def get_json(session, url, **kwargs):
    response = get_response(session, url, **kwargs)
    if response.text.lstrip().startswith("<"):
        raise ValueError("Received HTML instead of product JSON")
    try:
        data = response.json()
    except ValueError:
        raise ValueError("Invalid JSON; inspect the response format") from None
    if not isinstance(data, dict):
        raise ValueError("Expected a JSON object")
    return data

def variant_rows(product, currency="CAD"):
    if currency != "CAD":
        raise ValueError("Verify price scaling before adding another currency")

    variants = product.get("variants")
    if not product.get("id") or not isinstance(variants, list) or not variants:
        raise ValueError("Missing product ID or variants")

    captured = datetime.now(timezone.utc).isoformat()
    rows = []
    for variant in variants:
        raw_price = variant["price"]
        if type(raw_price) is not int:
            raise ValueError("Ajax price must be an integer")
        price = Decimal(raw_price) / Decimal(100)

        compare = variant.get("compare_at_price")
        if compare is not None and type(compare) is not int:
            raise ValueError("Ajax comparison price must be an integer or null")
        if type(variant.get("available")) is not bool:
            raise ValueError("Variant availability must be a Boolean")
        compare_price = None if compare is None else Decimal(compare) / 100

        rows.append({
            "product_id": str(product["id"]),
            "title": product["title"],
            "variant_id": str(variant["id"]),
            "variant": variant["title"],
            "sku": variant.get("sku") or "",
            "price": format(price, ".2f"),
            "currency": currency,
            "available": variant["available"],
            "compare_at_price": "" if compare_price is None else format(compare_price, ".2f"),
            "is_discounted": compare_price is not None and compare_price > price,
            "url": f"{BASE}/products/{product['handle']}?variant={variant['id']}",
            "captured_at": captured,
        })

    if len({row["variant_id"] for row in rows}) != len(rows):
        raise ValueError("Duplicate variant IDs")
    return rows

def save_rows(rows, stem):
    if not rows:
        raise ValueError("No rows to export")
    OUT.mkdir(exist_ok=True)
    (OUT / f"{stem}.json").write_text(
        json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    with (OUT / f"{stem}.csv").open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)


def main():
    OUT.mkdir(exist_ok=True)
    with make_session() as session:
        product = get_json(session, f"{BASE}/products/{HANDLE}.js")
        (OUT / "product_raw.json").write_text(
            json.dumps(product, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        rows = variant_rows(product)
        save_rows(rows, "puff_variants")
        print(f"Product: {product['title']}")
        print(f"Variants: {len(rows)} | Currency: CAD")
        for row in rows:
            print(row["variant"], row["sku"], row["price"], row["available"])
        print("Saved: results/puff_variants.csv and results/puff_variants.json")


if __name__ == "__main__":
    main()

python shopify_scraper.py

import json
import time
from shopify_scraper import BASE, OUT, get_json, make_session


def collect(session, max_pages=2, limit=2):
    products, seen = [], set()
    for page in range(1, max_pages + 1):
        data = get_json(session, f"{BASE}/products.json",
                        params={"limit": limit, "page": page})
        batch = data.get("products")
        if not isinstance(batch, list):
            raise ValueError("Missing products list")
        if not batch:
            break
        fresh = [p for p in batch if p["id"] not in seen]
        if not fresh:
            raise ValueError("Repeated page; stop and inspect pagination")
        for product in fresh:
            seen.add(product["id"])
            products.append(product)
        print(f"Page {page}: {len(batch)} products | Unique total: {len(seen)}")
        if page < max_pages:
            time.sleep(2)
    if not products:
        raise ValueError("No products returned; inspect the catalog response")
    return products


if __name__ == "__main__":
    with make_session() as session:
        products = collect(session)
        OUT.mkdir(exist_ok=True)
        (OUT / "catalog_sample.json").write_text(
            json.dumps(products, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        print("Saved: results/catalog_sample.json")
        print("Bounded sample only; not a complete store inventory")

import json
from pathlib import Path
from bs4 import BeautifulSoup
from shopify_scraper import BASE, HANDLE, get_response, make_session


def walk(value):
    if isinstance(value, dict):
        yield value
        for child in value.values():
            yield from walk(child)
    elif isinstance(value, list):
        for child in value:
            yield from walk(child)


def extract_offers(html):
    soup = BeautifulSoup(html, "html.parser")
    offers = []
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string or script.decode_contents())
        except json.JSONDecodeError:
            continue
        for item in walk(data):
            if item.get("@type") == "Offer" and "price" in item:
                offers.append(item)
    if not offers:
        raise ValueError("No price offers found; inspect this theme")
    return offers


if __name__ == "__main__":
    with make_session() as session:
        response = get_response(session, f"{BASE}/products/{HANDLE}")
        Path("results").mkdir(exist_ok=True)
        Path("results/product.html").write_text(response.text, encoding="utf-8")
        offers = extract_offers(response.text)
        Path("results/offers.json").write_text(
            json.dumps(offers, indent=2), encoding="utf-8"
        )
        print(f"HTML status: {response.status_code} | Offers: {len(offers)}")
        for offer in offers:
            print(offer["price"], offer.get("priceCurrency"), offer.get("availability"))

import xml.etree.ElementTree as ET
from urllib.parse import urlsplit
from shopify_scraper import BASE, OUT, get_response, make_session


def read_xml(session, url):
    target = (urlsplit(url).scheme, urlsplit(url).netloc)
    base = (urlsplit(BASE).scheme, urlsplit(BASE).netloc)
    if target != base:
        raise ValueError("Stop at a cross-domain sitemap")
    return ET.fromstring(get_response(session, url).content)


if __name__ == "__main__":
    ns = {"s": "http://www.sitemaps.org/schemas/sitemap/0.9"}
    with make_session() as session:
        index = read_xml(session, BASE + "/sitemap.xml")
        links = [node.text for node in index.findall("s:sitemap/s:loc", ns)]
        product_map = next((u for u in links if u and "sitemap_products" in u), None)
        if product_map is None:
            raise ValueError("No product sitemap found in this index")
        root = read_xml(session, product_map)
        urls = list(dict.fromkeys(
            node.text for node in root.findall("s:url/s:loc", ns)
            if node.text and "/products/" in urlsplit(node.text).path
            and (urlsplit(node.text).scheme, urlsplit(node.text).netloc)
            == (urlsplit(BASE).scheme, urlsplit(BASE).netloc)
        ))[:3]
        if not urls:
            raise ValueError("No in-scope product URLs found; inspect the sitemap")
        OUT.mkdir(exist_ok=True)
        (OUT / "sitemap_sample.txt").write_text("\n".join(urls), encoding="utf-8")
        print(f"First product sitemap: {len(urls)} sample URLs")
        for url in urls:
            print(url)

from getpass import getpass
from urllib.parse import quote
from shopify_scraper import BASE, HANDLE, get_json, make_session, variant_rows


def configure_proxy(session, host, port, username, password):
    if not host or any(c.isspace() or c in "/:@" for c in host):
        raise ValueError("Enter a gateway hostname only")
    if not str(port).isdigit() or not 1 <= int(port) <= 65535:
        raise ValueError("Invalid proxy port")
    if not username or not password:
        raise ValueError("Username and password are required for this authentication mode")
    proxy_url = f"http://{quote(username, safe='')}:{quote(password, safe='')}@{host}:{port}"
    session.proxies.update({"http": proxy_url, "https": proxy_url})


if __name__ == "__main__":
    host = input("Rola IP gateway hostname: ").strip()
    port = input("HTTP proxy port: ").strip()
    username = getpass("Full generated residential proxy username, including suffixes (hidden): ")
    password = getpass("Proxy password (hidden): ")
    try:
        with make_session() as session:
            configure_proxy(session, host, port, username, password)
            product = get_json(session, f"{BASE}/products/{HANDLE}.js")
            rows = variant_rows(product)
            print("Product JSON received:", product["title"], "| Validated variants:", len(rows))
    except RuntimeError as error:
        print("Request failed:", str(error))
        raise SystemExit(1)
    except Exception as error:
        print("Request failed:", type(error).__name__)
        raise SystemExit(1)