Lazadaスクレイピング入門：Pythonでの商品データ抽出、プロキシ設定、サービス比較

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "beautifulsoup4>=4.13,<5"

python3 -m venv .venv
.venv/bin/python -m pip install "beautifulsoup4>=4.13,<5"

import argparse
import csv
import hashlib
import json
import re
import sys
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
from uuid import uuid4
from bs4 import BeautifulSoup

MAX_BYTES = 2_000_000
MARKETS = {
    "SG": ("www.lazada.sg", "SGD"), "MY": ("www.lazada.com.my", "MYR"),
    "PH": ("www.lazada.com.ph", "PHP"), "TH": ("www.lazada.co.th", "THB"),
    "ID": ("www.lazada.co.id", "IDR"), "VN": ("www.lazada.vn", "VND")}
FIXTURE_PRODUCTS = [
    {"@type": "Product", "productID": "DEMO-ITEM-01", "sku": "DEMO-BLUE",
     "name": "Fictional travel mug - blue, 350 ml",
     "url": "https://catalog.example.invalid/mug-blue",
     "image": "https://images.example.invalid/mug-blue.jpg",
     "aggregateRating": {"ratingValue": "4.6", "bestRating": "5", "ratingCount": "12"},
     "offers": {"@type": "Offer", "price": "24.90", "priceCurrency": "SGD",
                "seller": {"name": "Demo Seller"},
                "availability": "https://schema.org/InStock"}},
    {"@type": "Product", "productID": "DEMO-ITEM-02", "sku": "DEMO-RED",
     "name": "Fictional travel mug - red, 350 ml",
     "url": "https://catalog.example.invalid/mug-red",
     "offers": {"@type": "Offer", "price": "26.50", "priceCurrency": "SGD",
                "seller": {"name": "Demo Seller"}}}]
FIXTURE = ('<html><script type="application/ld+json">' +
           json.dumps({"@context": "https://schema.org", "@graph": FIXTURE_PRODUCTS}) +
           '</script></html>').encode()

def fail(code):
    raise ValueError(code)

def typed(node, expected):
    value = node.get("@type", []) if isinstance(node, dict) else []
    return value == expected or (isinstance(value, list) and expected in value)

def text(value):
    return str(value).strip() if isinstance(value, (str, int, float, Decimal)) and not isinstance(value, bool) else ""

def safe_url(value, host=None):
    p = urlsplit(text(value))
    if p.scheme != "https" or not p.hostname or p.username or p.password or p.port not in (None, 443):
        fail("FIELD_URL")
    if host and p.hostname.lower() != host:
        fail("MARKET_HOST")
    return urlunsplit((p.scheme, p.netloc, p.path, "", ""))

def products_from_html(data):
    if len(data) > MAX_BYTES:
        fail("INPUT_SIZE")
    nodes = []
    for tag in BeautifulSoup(data, "html.parser").select('script[type="application/ld+json"]'):
        parsed = json.loads(tag.get_text(), parse_float=Decimal,
                            parse_constant=lambda value: fail("JSON_NUMBER"))
        roots = parsed if isinstance(parsed, list) else [parsed]
        for root in roots:
            if not isinstance(root, dict):
                continue
            if typed(root, "Product"):
                nodes.append(root)
            graph = root.get("@graph", [])
            if not isinstance(graph, list):
                fail("JSON_GRAPH")
            nodes.extend(node for node in graph if typed(node, "Product"))
    if not nodes:
        fail("NO_PRODUCT_ADAPTER")
    return nodes

def extract(data, market, kind, observed_at, run_id):
    host, currency = MARKETS[market]
    if kind == "synthetic_fixture":
        host = "catalog.example.invalid"
    rows, seen = [], set()
    for product in products_from_html(data):
        offer = product.get("offers")
        if isinstance(offer, list) and len(offer) == 1:
            offer = offer[0]
        if not typed(offer, "Offer"):
            fail("OFFER_ADAPTER")
        amount = text(offer.get("price"))
        if not re.fullmatch(r"\d+(?:\.\d+)?", amount) or Decimal(amount) <= 0:
            fail("FIELD_AMOUNT")
        if text(offer.get("priceCurrency")) != currency:
            fail("MARKET_CURRENCY")
        name = text(product.get("name"))
        url = safe_url(product.get("url"), host)
        if not name or url in seen:
            fail("IDENTITY_OR_DUPLICATE")
        seen.add(url)
        seller = offer.get("seller", {})
        seller = text(seller.get("name")) if isinstance(seller, dict) else ""
        availability = text(offer.get("availability"))
        stock = {"https://schema.org/InStock": "in_stock",
                 "https://schema.org/OutOfStock": "out_of_stock"}.get(availability, "unknown")
        rating = product.get("aggregateRating", {})
        if not isinstance(rating, dict):
            rating = {}
        image = product.get("image", "")
        if isinstance(image, list):
            image = image[0] if image else ""
        image = safe_url(image) if image else ""
        item_id, sku = text(product.get("productID")), text(product.get("sku"))
        review = ["purchase_conditions_unverified"]
        for missing, label in [(not item_id, "item_id_missing"), (not sku, "sku_missing"),
                               (not seller, "seller_missing"), (stock == "unknown", "availability_unknown")]:
            if missing:
                review.append(label)
        rows.append(dict(item_id=item_id, sku=sku, title=name, seller=seller,
                         display_price=str(Decimal(amount)), currency=currency,
                         availability=stock, listing_type="unknown", cross_border="unknown",
                         ship_from_country="", shipping_fee="", shipping_currency="",
                         lead_time_days="", platform_voucher="", seller_voucher="", bundle_offer="",
                         rating_value=text(rating.get("ratingValue")),
                         rating_scale=text(rating.get("bestRating")),
                         rating_count=text(rating.get("ratingCount")), image_url=image,
                         source_url=url, requested_market=market, source_kind=kind,
                         source_observed_at_utc=observed_at, run_id=run_id,
                         comparison_eligible=False, review_reason=";".join(review)))
    return rows

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--input", type=Path)
    p.add_argument("--approved", action="store_true")
    p.add_argument("--market", choices=MARKETS, default="SG")
    p.add_argument("--observed-at")
    p.add_argument("--out", type=Path, required=True)
    a = p.parse_args()
    log_path = a.out.with_suffix(".run.json")
    if a.out.exists() or log_path.exists():
        fail("OUTPUT_EXISTS")
    if a.input:
        if not a.approved or not a.observed_at:
            fail("REQUIRE_APPROVAL_AND_CAPTURE_TIME")
        stamp = datetime.fromisoformat(a.observed_at.replace("Z", "+00:00"))
        if stamp.tzinfo is None:
            fail("CAPTURE_TIME_ZONE")
        observed = stamp.astimezone(timezone.utc).isoformat()
        with a.input.open("rb") as handle:
            data = handle.read(MAX_BYTES + 1)
        kind = "approved_saved_html"
    else:
        if a.market != "SG":
            fail("FIXTURE_IS_SG_ONLY")
        data, kind, observed = FIXTURE, "synthetic_fixture", "2026-10-27T00:00:00+00:00"
    run_id = uuid4().hex
    rows = extract(data, a.market, kind, observed, run_id)
    log = dict(run_id=run_id, source_kind=kind, requested_market=a.market,
               source_observed_at_utc=observed,
               processed_at_utc=datetime.now(timezone.utc).isoformat(),
               input_sha256=hashlib.sha256(data).hexdigest(),
               records_exported=len(rows), proxy_tested=False)
    with log_path.open("x", encoding="utf-8") as handle:
        json.dump(log, handle, indent=2)
    with a.out.open("x", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print(f"Exported {len(rows)} records; inspect review_reason before analysis.")

if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        allowed = {"INPUT_SIZE", "FIELD_URL", "MARKET_HOST", "JSON_NUMBER", "JSON_GRAPH",
                   "NO_PRODUCT_ADAPTER", "OFFER_ADAPTER", "FIELD_AMOUNT", "MARKET_CURRENCY",
                   "IDENTITY_OR_DUPLICATE", "OUTPUT_EXISTS", "REQUIRE_APPROVAL_AND_CAPTURE_TIME",
                   "CAPTURE_TIME_ZONE", "FIXTURE_IS_SG_ONLY"}
        code = str(error) if isinstance(error, ValueError) and str(error) in allowed else "INPUT_OR_OUTPUT_ERROR"
        print("Stopped:", code, "- inspect input and configuration privately.", file=sys.stderr)
        sys.exit(1)

.\.venv\Scripts\python.exe lazada_extract.py --out lazada-demo.csv

.venv/bin/python lazada_extract.py --out lazada-demo.csv

.\.venv\Scripts\python.exe lazada_extract.py --input approved-product.html --approved --market SG --observed-at "2026-10-27T09:00:00+08:00" --out approved-sg.csv

.venv/bin/python lazada_extract.py --input approved-product.html --approved --market SG --observed-at "2026-10-27T09:00:00+08:00" --out approved-sg.csv

# Configuration excerpt inside your approved Playwright collector, not a standalone scraper.
import os
proxy = {
    "server": "http://" + os.environ["SOCKS5IO_HTTP_HOST"] + ":" + os.environ["SOCKS5IO_HTTP_PORT"],
    "username": os.environ["SOCKS5IO_USER"],
    "password": os.environ["SOCKS5IO_PASSWORD"],
}
# browser = playwright.chromium.launch(proxy=proxy)